22

我主要看到 Airflow 用于 ETL/投标数据相关的工作。我正在尝试将它用于业务工作流程，其中用户操作将来会触发一组相关任务。其中一些任务可能需要根据某些其他用户操作来清除（删除）。我认为处理此问题的最佳方法是通过动态任务 ID。我读到 Airflow 支持动态 dag id。因此，我创建了一个简单的 python 脚本，它将 DAG id 和任务 id 作为命令行参数。但是，我遇到了使它工作的问题。它给出了 dag_id not found 错误。有没有人试过这个？这是我在命令行上作为 python (python tmp.py 820 2016-08-24T22:50:00) 执行的脚本代码（称为 tmp.py）：

from __future__ import print_function
import os
import sys
import shutil
from datetime import date, datetime, timedelta
from airflow import DAG
from airflow.operators.bash_operator import BashOperator
execution = '2016-08-24T22:20:00'
if len(sys.argv) > 2  :
   dagid =  sys.argv[1]
   taskid = 'Activate' + sys.argv[1]
   execution = sys.argv[2]
else:
   dagid = 'DAGObjectId'
   taskid = 'Activate'
default_args = {'owner' : 'airflow', 'depends_on_past': False, 'start_date':date.today(), 'email': ['fake@fake.com'], 'email_on_failure': False, 'email_on_retry': False, 'retries': 1}
dag = DAG(dag_id = dagid,
       default_args=default_args,
       schedule_interval='@once',
      )
 globals()[dagid] = dag
task1 = BashOperator(
    task_id = taskid,
    bash_command='ls -l',
    dag=dag)

fakeTask = BashOperator(
    task_id = 'fakeTask',
    bash_command='sleep 5',
    retries = 3,
    dag=dag)
task1.set_upstream(fakeTask)

airflowcmd = "airflow run " + dagid + " " + taskid + "  " + execution
print("airflowcmd = " + airflowcmd)
os.system(airflowcmd)
4

2 回答 2

23

经过多次尝试和错误，我能够弄清楚这一点。希望它会帮助某人。以下是它的工作原理：您需要有一个迭代器或一个外部源（文件/数据库表）来通过模板动态生成 dags/task。您可以保持 dag 和任务名称静态，只需动态分配它们 id 以区分一个 dag 与另一个。你把这个 python 脚本放在 dags 文件夹中。当您启动气流调度程序时，它会在每个心跳上运行此脚本并将 DAG 写入数据库中的 dag 表。如果已经写入了一个 dag（唯一 dag id），它会简单地跳过它。调度器还会查看各个 DAG 的调度，以确定哪个 DAG 已准备好执行。如果 DAG 准备好执行，它会执行它并更新其状态。这是一个示例代码：

from airflow.operators import PythonOperator
from airflow.operators import BashOperator
from airflow.models import DAG
from datetime import datetime, timedelta
import sys
import time

dagid   = 'DA' + str(int(time.time()))
taskid  = 'TA' + str(int(time.time()))

input_file = '/home/directory/airflow/textfile_for_dagids_and_schedule'

def my_sleeping_function(random_base):
    '''This is a function that will run within the DAG execution'''
    time.sleep(random_base)

def_args = {
    'owner': 'airflow',
    'depends_on_past': False,
    'start_date': datetime.now(), 'email_on_failure': False,                
    'retries': 1, 'retry_delay': timedelta(minutes=2)
}
with open(input_file,'r') as f:
    for line in f:
        args = line.strip().split(',')
    if len(args) < 6:
        continue
    dagid = 'DAA' + args[0]
    taskid = 'TAA' + args[0]
    yyyy    = int(args[1])
    mm      = int(args[2])
    dd      = int(args[3])
    hh      = int(args[4])
    mins    = int(args[5])
    ss      = int(args[6])
    dag = DAG(
        dag_id=dagid, default_args=def_args,
        schedule_interval='@once', start_date=datetime(yyyy,mm,dd,hh,mins,ss)
        )

    myBashTask = BashOperator(
        task_id=taskid,
        bash_command='python /home/directory/airflow/sendemail.py',
        dag=dag)

    task2id = taskid + '-X'

    task_sleep = PythonOperator(
        task_id=task2id,
        python_callable=my_sleeping_function,
        op_kwargs={'random_base': 10},
        dag=dag)

    task_sleep.set_upstream(myBashTask)

f.close()
于 2016-09-23T18:04:05.413 回答
15

从如何动态创建 DAG？：

Airflow 在你的 [sic] DAGS_FOLDER 中查找在其全局命名空间中包含 DAG 对象的模块，并将它在 DagBag 中找到的对象添加。知道了这一点，我们所需要的就是一种在全局命名空间中动态分配变量的方法，这很容易在 python 中使用标准库的 globals() 函数来完成，它的行为就像一个简单的字典。

for i in range(10):
    dag_id = 'foo_{}'.format(i)
    globals()[dag_id] = DAG(dag_id)
    # or better, call a function that returns a DAG object!
于 2017-05-23T06:48:11.117 回答