有没有办法在Airflow中创建动态工作流程

Question

所以我有任务A将一些未知数量的文件复制到一个文件夹中。任务B在文件夹中的每个文件上运行。我无法预先知道文件的数量，因为它们不断变化。有没有办法让这项工作在气流中完成。

spans = os.listdir('/home/abc/tmpFolder')
counter = 0
for s in spans:
    src_path = '/home/abc/tmpFolder' + s
    dst_path = "tmp/" + s
    counter += 1
    run_this = \
        FileToGoogleCloudStorageOperator(
            task_id='gcp_task_' + str(counter),
            src=src_path,
            dst=dst_path,
            bucket='gcpBucket',
            google_cloud_storage_conn_id='gcp',
            mime_type='text/plain',
            dag=dag
        )
    dummy_operator_two.set_downstream(run_this)

我得到目录中所有文件的名称，然后为它们运行运算符，但气流不能正常工作，因为它需要事先知道数字。

Answer 1

我不希望Airflow在DagRun处于活动状态时修改DAG，所以我不会赌钱获取文件，然后在同一个DAG中添加任务。话虽这么说，Airflow每隔几秒就会重新生成DAG。您可以拥有一个获取文件的DAG和另一个处理这些文件的DAG。获取文件后，首先DAG必须等待一分钟，以确保Airflow注意到，然后用TriggerDagRunOperator启动第二个DAG。

DAG1：

def wait_a_minute():
    time.sleep(60)

get_files = DummyOperator(dag=dag, task_id='get_files')
give_airflow_time_to_rebuild_DAG2 = PythonOperator(dag=dag, task_id='give_airflow_time_to_rebuild_DAG2', python_callable=wait_a_minute)
trigger_DAG2 = TriggerDagRunOperator(dag=dag, task_id='trigger_DAG2', trigger_dag_id='DAG2', execution_date='{{ ds }}')

get_files >> give_airflow_time_to_rebuild_DAG2 >> trigger_DAG2

DAG2：

pre_process = DummyOperator(dag=dag, task_id='pre_process')
post_process = DummyOperator(dag=dag, task_id='post_process')

files = get_files_to_process()

for file in files:
    process = DummyOperator(dag=dag, task_id=f'process_{file}')
    pre_process >> process >> post_process

比解决方案更糟糕，但这样的事情应该有效。但是外部触发器和动态任务存在问题。当我不得不使用depends_on_past=True时，我经常偶然发现调度问题。

Answer 2

我通过制作单独的管道而不是单独的任务来完成这类工作。

Answer 3

您是否尝试过使用glob module并修改管道以处理给定目录中的所有文件？

Answer 4

关于my blog post on Creating a Dynamic Workflow using Apache Airflow，您可以测试以下代码：

import airflow
from airflow import DAG
from airflow.operators.dummy_operator import DummyOperator
from airflow.operators.bash_operator import BashOperator
from airflow.contrib.operators.file_to_gcs import FileToGoogleCloudStorageOperator
from datetime import datetime,timedelta
from os import listdir

default_args = {
  'owner': 'test',
  'depends_on_past': False,
  'start_date': airflow.utils.dates.days_ago(1),
  'retries': 1,
  'retry_delay': timedelta(minutes=5)
}

dag = DAG('dynamic',
  default_args=default_args,
  description='Dynamic DAG',
  schedule_interval=timedelta(days=1))

copy_files = BashOperator(task_id='cp_files',
   depends_on_past=False,
   bash_command='cp /tmp/filetocopy/* /tmp/filetoprocess',
   dag=dag)

start = DummyOperator(task_id='start',
                     dag=dag)
end = DummyOperator(task_id='end',
                   dag=dag)

start >> copy_files

spans = listdir('/tmp/filetoprocess')
counter = 1

for s in spans:
  src_path = '/tmp/filetoprocess/' + s
  dst_path = "/tmp/dest/" + s
  counter += 1
  task = FileToGoogleCloudStorageOperator(
            task_id='gcp_task_' + str(counter),
            src=src_path,
            dst=dst_path,
            bucket='gcpBucket',
            google_cloud_storage_conn_id='gcp',
            mime_type='text/plain',
            dag=dag)
  task.set_upstream(copy_files)
  task.set_downstream(end)

使用此代码，您需要已经有一些文件（您还可以在/tmp/filetoprocess文件夹中创建一个Python函数，用于检查是否存在某些文件，否则创建DummyOperator以使整个工作流程正常工作）;否则，Airflow调度程序将产生问题以生成正确的DAG。

我已经使用新的Apache Airflow版本（v.1.10）测试它，它似乎完美无缺。

Dynamic tasks on Airflow DAG