我正在尝试为DAG编写脚本。我已经运行了docker-airflow,并且通过localhost在浏览器上运行良好。但是,每当我尝试在vscode中打开一个dag文件时,它都会显示导入错误。 我已经安装和设置气流使用docker和激活docker使用芹菜执行器。 Docker Container for airflow正在运行。 from airflow import DAG
from airflow.operators.bash_operator import BashOperator ?
我已经用docker-compose设置了气流,如下所述。https://airflow.apache.org/docs/apache-airflow/stable/start/docker.html 还有一个必须执行docker命令的气流任务,比如 BashOperator(
task_id='my',
bash_command="""
docker run ..............
""",
dag=dag,
) 这意味着Docker包需要in the airflow do
我正在运行空气流量2.1.4使用码头-合成和芹菜执行器。到目前为止,我已经能够从芹菜工作容器启动和运行简单的DockerOperator任务,但是现在当我尝试将一个目录从共享驱动器挂载到任务容器时,我得到了一个错误(下面的日志文件)。如果我不定义挂载参数,Dag工作得很好。因此,我猜有些信息或特权不会传递给由芹菜工人容器创建的容器。有什么建议吗?接下来要尝试什么?
DAG档案:
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetim
我对气流很陌生。我能够跟踪一个视频并创建docker-compose yml文件、Dockerfile和一个dag文件。我可以查看我的守护进程并运行它。在我的脚本中,我试图打开一个文本文件(.txt),但是我得到了以下错误:FileNotFoundError: \[Errno 2\] No such file or directory。
我的文本文件在正确的位置。脚本运行在本地python环境上。我不知道当我在气流中运行时,它为什么会显示为一个错误。
我的docker-compose.yml、Dockerfile和dag文件如下所示。任何帮助我都会感激!谢谢!
docker-compose.y
在DAG中,我使用的是一个DockerOperator,在其中我需要挂载一个临时目录来存储一些数据。容器必须为这个临时目录在主机上使用特定的路径,所以我尝试使用DockerOperator的“DockerOperator”参数,但这是行不通的。
考虑以下DAG示例:
from airflow import DAG
from airflow.providers.docker.operators.docker import DockerOperator
from datetime import datetime
with DAG(dag_id="test_v1",
我有一个简单的示例python脚本,我希望Airflow在计划的基础上运行。我可以为Airflow创建容器,并在该容器中成功安装Docker服务器。我可以手动访问CLI并"docker run“示例的容器,它就可以工作了。当我从本地主机管理员尝试它时,它只显示“运行”永远。在CLI中检查"docker ps“显示它还没有启动容器。我确信我错过了一些简单的东西,因为这里没有太多复杂的东西。请帮帮忙!?
以下是python脚本:
import time as ti
def main():
print('TEST')
ti.sleep(120)
我试图使用码头操作员自动执行一些脚本使用气流。
气流版本:apache-airflow==1.10.12
我想要做的是“复制”我的项目的所有文件(文件夹和文件)到容器使用这段代码。
以下文件ml-intermediate.py位于此目录~/airflow/dags/ml-intermediate.py中
"""
Template to convert a Ploomber DAG to Airflow
"""
from airflow import DAG
from airflow.operators.bash_operator import B
我最近开始使用Docker airflow (puckel/docker-airflow),这让我做了噩梦。
我想使用BashOperator运行一个bash脚本。但是当它运行时,它找不到脚本位置。
这是我的代码:
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
from datetime import datetime, timedelta
import os
default_args = {
"owner": "airflow",
我在码头使用气流和PostgreSQL。
因此,我在端口5433上建立了一个PostgreSQL数据库。集装箱(384 Ea7b6efb)。这是我有我的数据,我想拿我的dag在气流中。
码头工人ps
CONTAINER ID IMAGE COMMAND CREATED STATUS
PORTS NAMES
384eaa7b6efb postgres "docker-entrypoint.s…
上下文
我正在运行气流,并试图运行一个概念的证明码头集装箱使用气流的。我部署到库伯内特斯(EKS),但还没有使用库伯内特斯执行者。考虑到我是在运行豆荚,通过使用DockerOperator,我将在码头上运行(据我理解) Docker。
每当我运行任务时,我都会收到错误:ERROR - Error while fetching server API version。错误既发生在docker-compose上,也发生在EKS (kubernetes)上。
我目前的状况
Thi是我的气流Dockerfile的样子:
FROM apache/airflow:1.10.14-python3.8
# Us
当在容器内的气流任务中使用熊猫pd.read_excel()时,我会看到下面的openpyxl错误。我尝试过使用诗歌安装openpyxl,甚至在dockerfile中使用pip,但没有成功。
File "/home/airflow/.local/lib/python3.8/site-packages/pandas/io/excel/_openpyxl.py", line 521, in __init__
import_optional_dependency("openpyxl")
File "/home/airflow/.local/li
我使用CLI创建了一个自定义包(使用Click构建)。这个包可以做两件事:运行预处理和运行机器学习模型。我创建了此客户包的Docker镜像,并将其推送到AWS (ECR)上的私有注册表。 现在我想用Airflow运行这个容器,我想在一个EC2实例上运行它。我正在用docker-compose运行它。 在本例中,我将只关注一项任务:运行容器进行预处理。 然而,现在我得到了‘上游失败’的t2。 from datetime import timedelta
import airflow
from airflow import DAG
from airflow.operators.bash_oper
我正在尝试运行一个简单的python脚本中的一个码头运行命令调度与气流。
我遵循了这里的说明,。
我的.env文件:
AIRFLOW_UID=1000
AIRFLOW_GID=0
而docker-compose.yaml是基于默认的。我必须添加- /var/run/docker.sock:/var/run/docker.sock作为一个额外的卷来运行码头内的码头。
我的守护进程配置如下:
""" this is an example dag """
from datetime import timedelta
from airflow impo
我正在使用气流稳定舵图和Kubernetes Executor,新的pod正在为dag调度,但它与dag_id的故障无法找到问题。我正在使用git-sync来获取dags。下面是错误和kubernetes配置值。有人能帮我解决这个问题吗?
错误:
[2020-07-01 23:18:36,939] {__init__.py:51} INFO - Using executor LocalExecutor
[2020-07-01 23:18:36,940] {dagbag.py:396} INFO - Filling up the DagBag from /opt/airflow/dags/dag
我正在使用Windows机器,并且已经创建了用于气流的容器。我可以通过DAG读取本地文件系统上的数据,但无法将数据写入文件。我还尝试了给出完整路径,也尝试了不同的运算符: Python和Bash,但仍然不起作用。DAG成功了,没有任何失败的迹象。注: /opt/airflow :是$AIRFLOW_HOME路径
可能的原因是什么?
以下是一段代码:
from airflow import DAG
from datetime import datetime
from airflow.operators.python import PythonOperator
from airflow.opera