我在dags目录中有两个文件- dag_1.py和dag_2.py
dag_1.py创建静态DAG，dag_2.py根据外部json文件在某个位置创建动态DAG。
静态DAG (由dag_1.py创建)包含一个稍后阶段的任务，该任务为dag_2.py和动态DAG生成一些这些输入json文件。
过去，在不使用DAG序列化的情况下，气流1.x版本可以很好地工作。但随着气流2.0DAG系列化已成为强制性的。有时，当生成动态DAG时，我会在Scheduler中得到以下异常-
[2021-01-02 06:17:39,493] {scheduler_job.py:1293} ERROR - Except
浏览 11提问于2021-01-05得票数 4
回答已采纳
我有一个简单的dag：(3个python操作) default_args = {
'owner': 'default_user',
'start_date': days_ago(2),
'retries': 5,
'retry_delay': timedelta(minutes=1),
}
dag = DAG(
'basic_dag',
default_args=default_args,
schedule_interval=timedel
浏览 43提问于2020-11-24得票数 2
回答已采纳
我想登录到Apache Airflow，但是当我登录时，我得到了一个错误。但我已经在Ubuntu中创建了用户。我不知道问题在哪里，以及它的解决方案。有人知道怎么解决这个问题吗？我从Windows10操作系统在Ubuntu中安装了apache airflow 2.1.3版本，我真的不使用kubernetes或docker，我的SQLAlchemy版本是1.3.24
Something bad has happened.
Please consider letting us know by creating a bug report using GitHub.
Python version: 3
浏览 7提问于2021-09-12得票数 0
我正在从一个列表中动态地创建on_failure_callback，并且想要向其中一个任务添加dags。我尝试了下面的代码，但是回调似乎没有被执行。 dag_ids = ['dag_a', 'dag_b', 'dag_c']
for dag_id in dag_ids:
def failure_callback():
logging.info('Inside failure callback for {}'.format(dag_id))
def python_callable(dag_id
浏览 52提问于2019-02-18得票数 0
回答已采纳
目前，我有一个DAG，由4个操作符组成，如下所示：
with DAG('dag', default_args=args, schedule_interval=schedule_interval, catchup=True) as dag:
main_dag = PythonOperator(
task_id='1',
python_callable=func,
provide_context=True,
dag=dag)
run_after_main_dag_1 = PythonOperator(
task_id=
浏览 0提问于2018-10-19得票数 2
我试图理解如何在Apache airflow中创建动态dags，因为我需要它来在我的项目中创建动态dags。 下面是iam的链接：Dynamic DAG creation in Apache airflow 下面是用于创建示例hello world动态DAGs的代码块(基于输入参数的动态DAG创建)。 from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
def create_dag(dag_id,
浏览 12提问于2019-06-12得票数 1
dag@Arokh:~$ source /home/dag/.bashrc
dag@Arokh:~$ echo $PATH
/home/dag/.local/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin:/usr/games:/usr/local/games:/snap/bin
dag@Arokh:~$ python3 /home/dag/.local/bin/facemorpher --version
Face Morpher 1.0
dag@Arokh:~$ python3 facemorpher --ve
浏览 5提问于2021-03-18得票数 0
我正在尝试使用ExternalTaskSensor，它被困于戳另一个DAG的任务，这个任务已经成功地完成了。
在这里，第一个DAG "a“完成它的任务，然后通过ExternalTaskSensor触发第二个DAG "b”。相反，它被困在a.first_task上。
第一次DAG：
import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
dag = DAG(
dag_id='a',
default_ar
浏览 8提问于2017-10-18得票数 21
回答已采纳
我试图在气流中运行一个简单的dag来执行python文件，它正在抛出错误，无法打开文件‘/User/.’。
下面是我正在使用的脚本。
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime,ti
浏览 9提问于2021-02-25得票数 2
回答已采纳
我想要获取执行的DAG的run_id，并将其作为DAG本身的一个变量。有这样的函数吗？ with DAG(
dag_id='question',
...
) as dag:
question = PythonOperator(
task_id='question_python',
python_callable=question_python,
op_kwargs={run_id:#here },
)
浏览 128提问于2021-11-20得票数 1
回答已采纳
我正在尝试独立和并行地运行任务， 我看起来是这样的： ---> patternA ---> file1a
---> file2a
---> file3a
sensor ---> move_csv ---> result_mv
浏览 24提问于2020-03-19得票数 0
我对气流很陌生。我有一个DAG.py，其中我使用一个BashOperator运行一个python脚本。
我想使用一个配置文件在这个python脚本中传递一些日期参数。我看到可以在UI上通过配置触发DAG：
我不知道如何在我的DAG.py中读取它并将它传递给python脚本作为参数使用。
我的DAG.py看起来是这样的：
from airflow import DAG
from airflow.operators.bash import BashOperator
dag = DAG(
'Sample_DAG',
description=&#
浏览 2提问于2022-11-15得票数 0
回答已采纳
我正试图从另一个人身上触发另一个人。我也在使用TriggerDagRunOperator。
我有以下两条线索。
Dag 1:
from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.dagrun_operator import TriggerDagRunOper
浏览 4提问于2020-01-06得票数 0
我试图通过将现有的气流任务分配给不同的dags来重用它。
def create_new_task_for_dag(task: BaseOperator,
dag: models.DAG) -> BaseOperator:
"""Create a deep copy of given task and associate it with given dag
"""
new_task = copy.deepcopy(task)
new_task.dag = dag
浏览 4提问于2020-08-22得票数 2
我有几个python文件，我目前正在使用BashOperator执行它们。这使我可以很容易地灵活地选择python虚拟环境。
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
default_args = {
'owner': 'airflow',
'depends_on_past': False,
...}
dag = DAG('python_tasks', default_args=defa
浏览 1提问于2018-04-10得票数 19
1回答
我有一个打包的DAG，其中我在子文件夹中存储我的模块。我正在使用PythonVirtualenvOperator，并希望从虚拟环境中访问此模块。 文件夹系统- dags/
packaged_dag.zip/
dag.py
package/
my_module.py
__init__.py dag.py from airflow.operators.python_operator import PythonVirtualenvOperator
def my_function(**kwargs):
浏览 15提问于2021-01-11得票数 1
我想将参数传递到气流DAG中，并在python函数中使用它们。我可以将参数使用到bash操作符中，但是我找不到任何引用来将它们用作python函数。
from airflow import DAG
from airflow.operators.bash_operator import BashOperator from airflow.operators.python_operator import PythonOperator from airflow.utils.dates import days_ago
#Define DAG
dag = DAG("test_backup
浏览 1提问于2021-06-03得票数 0
回答已采纳
我正在使用airflow，我想将任务1的函数的输出传递给任务2。 def create_dag(dag_id,
schedule,
default_args):
def getData(**kwargs):
res = requests.post('https://dummyURL')
return res.json())
def sendAlert(**kwargs):
requests.post('https://dummyURL'
浏览 18提问于2020-07-24得票数 0
我有一个如下所示的守护程序: ingest_excel.py：
from __future__ import print_function
import time
from builtins import range
from datetime import timedelta
from pprint import pprint
import airflow
from airflow.models import DAG
#from airflow.operators.bash_operator import BashOperator
from airflow.operators.pytho
浏览 1提问于2018-11-26得票数 1
回答已采纳
Python项目正在运行多个py脚本。是否可能不是从DAG调用其他python脚本，而是从单个python脚本调用其他python脚本，即通过气流中的PythonOperator或BashOperator调用DAG中提到的部分脚本。
例如，我想使用test.py调用其他python脚本
curl = BashOperator(
task_id='multiplescripts',
bash_command="test.py",
dag=dag)
浏览 12提问于2022-06-21得票数 0
3回答
关于“动态任务”的其他问题似乎解决了在计划或设计时动态构建DAG的问题。我对在执行过程中将任务动态添加到DAG很感兴趣。
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from datetime import datetime
dag = DAG('test_dag', description='a test',
浏览 2提问于2018-02-03得票数 21
回答已采纳
2回答
我是一个新的气流，想运行一个循环的任务，但我面临的循环错误。
from airflow import DAG
from airflow.contrib.operators.ssh_operator import SSHOperator
from airflow.operators.dummy import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.contrib.hooks.ssh_hook import SSHHook
from datetime import
浏览 6提问于2021-04-15得票数 0
回答已采纳
1回答
我正在编写一个使用PythonOperator的气流数据采集卡。它需要一个"python_callable“，它似乎是一个函数。我称这个Python代码为dag-trans.py
我有一个Python代码，它通过熊猫(transf.py)进行一些转换
我需要dag-trans.py任务作为"python_callable“参数来执行transf-py。
with DAG(dag_id='dag_transformation',
default_args=default_args,
catchup=False,
浏览 2提问于2021-09-03得票数 2
气流2：我已经将xcom从taskA中推了出来，并且在subdag taskB中拉出了xcom。我一直无法拔出必要的xcom。我在想一个人会怎么做。
我能够在气流1.10中使用相同的代码来提取必要的xcom (另外，我传递的是provide_context=True，是在气流2中分解的)，但是它在气流2中的失败与错误消息如下：
jinja2.exceptions.UndefinedError: 'DAG_ID' is undefined
代码如下：
DAG_ID = 'test_dag_name'
dag = DAG(
DAG_ID,
defa
浏览 4提问于2022-02-10得票数 0
回答已采纳
我对现有的工作流做了一个小小的修改，它破坏了气流。以下是代码：
dag_name = platform + "_" + report['table']
dag = DAG(
dag_name,
catchup=True,
default_args=default_args,
schedule_interval=report['schedule']
)
with dag:
trigger_report = PythonOperator(
task_id=dag.dag_id + '
浏览 0提问于2019-06-16得票数 4
回答已采纳
我有一个具有以下结构的气流DAG。
所有以"check*“开头的函数都是BranchPythonOperator，而函数exceptionControl是接收每个错误以处理它们的ExecuteDagRunOperator。
这是DAG配置：
checkCloudFunctions = BranchPythonOperator(
task_id='checkCloudFunctions',
python_callable=check_cloud_functions,
provide_context=True,
dag=dag)
浏览 0提问于2018-10-23得票数 2
我已经设置了airflow，并且正在使用以下vscode调试配置运行DAG：
{
"version": "0.2.0",
"configurations": [
{
"name": "Python: Current File",
"type": "python",
"request": "launch",
"progr
浏览 2提问于2020-12-17得票数 3
我使用的是airflow 1.10.3。我正在尝试创建一个可以多次实例化的类，以便为我生成DAG。下面是一个简化的示例： from airflow import DAG
from airflow.operators.python_operator import PythonOperator
class DagClass:
def __init__(self, dag_id):
self.dag_id = dag_id
def test_function():
print("test 1")
def test_
浏览 21提问于2019-09-18得票数 2
我需要计算python运算符中的值，并在其他操作符中使用它，如下面所示，我得到了火花提交和电子邮件操作符/的"dag_var不存在“。
我将dag_var声明为python可调用的全局变量。但我不能进入其他运营商。
def get_dag_var(ds, **kwargs):
global dag_var
dag_var = kwargs['dag_run'].run_id
with DAG(
dag_id='sample',
schedule_interval=None, # executes at 6 AM UTC
浏览 6提问于2022-04-12得票数 0
回答已采纳
在下面显示的DAG中，无论任务b和c是成功还是失败，我都想执行任务d，但是对于任务e，如果任务b，c和d是成功的，那么只应该触发它。DAG Image 写在下面的代码，但它不工作： with DAG(dag_id='test_dag', default_args=args, schedule_interval=None) as dag:
t_test = PythonOperator(task_id='test', python_callable=test)
t_a = PythonOperator(task_id='a', pyt
浏览 29提问于2021-06-15得票数 0
回答已采纳
目前，我有一个带有Python运算符的气流中的DAG，相关的python可以这样调用：
def push_xcom(**kwargs):
ti = kwargs["ti"]
ti.xcom_push(key=key, value=value)
xcom_opr = PythonOperator(
task_id='xcom_opr',
python_callable=push_xcom,
dag=dag
)
这个dag的目标是更新在气流中定义的其他DAG的xcom变量。这不可能吗？我找不
浏览 2提问于2019-12-04得票数 2
回答已采纳
我正在使用TriggerDagRunOperator，以便一个控制器DAG可以触发一个目标DAG。但是，一旦控制器DAG触发了目标DAG，目标DAG就会切换到"running"，但它的任何任务都不会被调度。我希望一旦控制器DAG触发目标DAG，就安排目标DAG的任务。 # Controller DAG's callable
def conditionally_trigger(context, dag_run_object):
condition_param = context['params']['condition_param'
浏览 10提问于2019-06-11得票数 0
我正在尝试调用一个python操作符，它位于一个函数中，使用另一个python操作符。似乎我错过了什么，有人能帮我找出我错过了什么吗？
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.python_operator import PythonOperator
from airflow.operators.bash_operator import BashOperator
from datetime import datetime,
浏览 2提问于2020-10-10得票数 5
我正在尝试创建一个动态的工作流程。我得到了这个：
我尝试使用BashOperator动态创建任务(它调用python脚本)
我的达格：
import datetime as dt
from airflow import DAG
import shutil
import os
from airflow.operators.bash_operator import BashOperator
from airflow.operators.python_operator import PythonOperator, BranchPythonOperator
from airflow.operat
浏览 3提问于2020-03-11得票数 0
我们已经转移到AirFlow 1.10.2来解决CPU的使用问题，好的是我们已经解决了这个问题，在我们的环境中得到了解决。但是，我们已经观察到，DAG的任务正在提交并显示在AirFlow仪表板上运行，但在实际执行发生之后，它们似乎会在队列中停留大约60秒。请注意，对于我们的用例实现
AirFlow不依赖于时间，即它们不是‘**计划中的’**，而是通过python代码触发的。
AirFlow v1.10.2被用作一个独立的分期付款executor = LocalExecutor。
python代码监视到达的任何文件的目录。它观察到，对于任何文件，代码都会触发AirFlow DAG
浏览 0提问于2019-03-09得票数 3
1回答
我有以下DAG，它使用一个专用于数据预处理例程的类来执行不同的方法：
from datetime import datetime
import os
import sys
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
import ds_dependencies
SCRIPT_PATH = os.getenv('MARKETING_PREPROC_PATH')
if SCRIPT_PATH:
sys.path.insert(0
浏览 2提问于2017-09-08得票数 4
回答已采纳
在我们的大数据项目中，有3000个表需要加载，所有这些表都应该由气流中的一个单独的DAG来处理。
在我们的解决方案中，单个python文件生成每种类型的表加载器，因此它们可以通过基于事件的REST通过云函数单独触发。因此，我们使用以下方法生成我们的DAG：
用于DAG生成器逻辑scd2的气流变量生成表类型的表名列表:插入追加、截断加载、scd1、
表加载器DAG的特定运算符使用的RawToRawRC_TableN 气流变量，例如:用于操作器处理的RR_TableN = {} // python = {} / python用于操作处理RawToCuration
用户定义宏:在任务定义之间，我们尝
浏览 3提问于2021-11-06得票数 1
回答已采纳
我有一个通过airflow API在DAG AIRFLOW中调用python脚本的用例。这个python脚本将通过JSON接收参数。我可以在shell上运行python，它会返回预期的结果。 我的宣传短片如下- curl -X POST -H "Cache-Control: no-cache" -H "Content-Type: application/json" http://localhost:8080/api/experimental/dags/DAG-3/dag_runs -d '{"conf":"{\"host
浏览 72提问于2020-07-18得票数 0
回答已采纳
我可以看到，在气流2.0之前，可以通过传递一个TriggerDagRunOperator选项来使用带有条件的python_callable：
def foo(context, dag_run_obj):
if True:
return dag_run_obj
dag = DAG(dag_id='test_trigger_dag_run_for_Sid',
default_args={"owner" : "me",
"start_date"
浏览 2提问于2021-06-16得票数 0
回答已采纳
2回答
我试图在气流DAG中编写Python操作符，并将某些参数传递给Python可调用程序。
我的代码如下所示。
def my_sleeping_function(threshold):
print(threshold)
fmfdependency = PythonOperator(
task_id='poke_check',
python_callable=my_sleeping_function,
provide_context=True,
op_kwargs={'threshold': 100},
dag=dag)
end
浏览 0提问于2019-02-15得票数 9
回答已采纳
1回答
我使用Airflow 1.10.9 (基于puckel/docker-airflow码头映像)通过BashOperator在DAG中运行多个Python。日志当前被写入/usr/local/airflow/logs。
是否可以将气流配置为
/home/foo/logsThe 还将日志写入另一个目录，比如日志应该只包含来自python脚本的stdout，日志应该以以下目录/文件名格式存储：
如果可能的话，/home/foo/logs/execution-date-dag-id-task-id.logRetries应该附加到同一个.log文件中。否则，我们可以使用命名约定：
/home/foo/l
浏览 0提问于2020-05-12得票数 2
我在Airflow 1.10上有一个奇怪的bug。我想试着发送电子邮件和微软团队的通知。我做了一个小的哑巴DAG来试一下。一切运行正常，但我连续收到两个通知。团队中的2封电子邮件和2条消息。 我在团队中使用了这个：https://github.com/mendhak/Airflow-MS-Teams-Operator 这里是Dags： from datetime import datetime
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.ope
浏览 34提问于2020-06-17得票数 4
1回答
我按照中的说明动态创建DAGs，通过变量k修改要创建的dags的数量
from datetime import datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
def create_dag(dag_id,
schedule,
dag_number,
default_args):
def hello_world_py(*args):
prin
浏览 16提问于2021-02-20得票数 2
1回答
我想动态地从另一个dag中调用DAG，例如，当操作符将触发这个父DAG时，它应该会生成新的DAG。
以下代码运行时没有错误，但DAG未创建。
我看不出原因。
def generate_dag():
def do_something():
print('something')
default_args = {
'schedule_interval': '@hourly',
'start_date': datetime(2022, 1, 1),
浏览 8提问于2022-03-24得票数 0
你好，我使用的是气流，这里是我试图解决的场景，我想在函数运行后动态创建DAG。
try:
import os
import sys
from datetime import timedelta,datetime
from airflow import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.t
浏览 7提问于2021-03-31得票数 0
1回答
我有一个函数，它将从带有dag信任的数据库中生成dags (我知道，这样做很昂贵)。问题是，只有当我在我定义的文件中调用这个函数时，它才会生成dags，如果我导入另一个文件并执行它，它就不会生成我的dags。
例：
def generate_dags_dinamically():
dags = get_dag_configs()
# this 'dags' variable, contains some configs for generating the dags
for dag in dags:
# Defines dag and
浏览 2提问于2022-11-18得票数 0
回答已采纳
我正在用Python3.8编写一些气流DAG代码，但是有一个缩进错误，我无法弄清楚。我使用VScode作为IDE
以下是代码：
from airflow_env import DAG
from datetime import datetime
with DAG(
dag_id='user-processing',
schedule_interval='@daily',
start_date=datetime(2022, 1, 1)) as dag:
在终端上，错误如下：unexpected EOF while parsing，但在expected
浏览 13提问于2022-03-23得票数 -2
在我的airflow dag中，我有一个ecs_operator任务，后跟python操作员任务。我想使用airflow的xcom功能将一些消息从ECS任务推送到python任务。我尝试了do_xcom_push=True选项，但没有结果。在下面的示例dag中找到。 dag = DAG(
dag_name, default_args=default_args, schedule_interval=None)
start = DummyOperator(task_id = 'start'
,dag =dag)
end = Dummy
浏览 10提问于2019-12-17得票数 2
在Airflow中，我试图在airflow中使用jinja模板，但问题是它没有被解析，而是被当作字符串处理。请参阅我的代码
from datetime import datetime
from airflow.operators.python_operator import PythonOperator
from airflow.models import DAG
def test_method(dag,network_id,schema_name):
print "Schema_name in test_method", schema_name
third
浏览 1提问于2018-11-26得票数 2
我有一个达格，每小时安排一次。假设凌晨1:00，凌晨2:00，凌晨3:00假设02:00被选中，但是如果01:00am dag运行仍在进行中，则需要取消02:00am实例。
我在尝试这个密码。
local_tz = pendulum.timezone("America/Chicago")
default_args = {
'owner': 'airflow',
'depends_on_past': True,
'start_date': datetime(2020, 11, 15, tzi
浏览 15提问于2020-11-17得票数 0
回答已采纳
云服务器
ICP备案
对象存储
腾讯会议
实时音视频
腾讯云开发者
