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

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

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) 

我正在获取目录中所有文件的名称，然后为它们运行运算符，但 Airflow 无法以这种方式工作，因为它需要事先知道编号。

请您参考如下方法:

我不希望 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 时，我通常会遇到调度程序问题。