在 Cloud Composer 中将同一运算符作为多个任务执行

百变鹏仔 1天前 #Python

文章标签多个

问题内容

我在 airflow 中有一个使用 cloud composer 执行的 pythonoperator：

with DAG(    dag_id = config['dag_id'],    schedule_interval = config['schedule_interval'],    default_args = default_args    ) as dag:        generate_data_task = PythonOperator(        task_id = 'generate_dummy_data',        python_callable = generate_data,        dag = dag    )

generate_data() 函数将随机生成的唯一命名的 csv 文件写入包含一些数据的存储桶中。按原样执行效果很好，但我想并行多次执行同一任务。如果我指定并行执行 10 次，我预计会在存储桶中写入 10 个文件。我尝试过并发和task_concurrency，但得到了相同的结果。

这可以在 cloud composer 之上使用 airflow 实现吗？

正确答案

使用动态任务映射：

generate_data_task = PythonOperator.partial(        task_id = 'generate_dummy_data',        python_callable = generate_data,        dag = dag    ).expand(op_args=[[]] * 10)

文章推荐

在 Cloud Composer 中将同一运算符作为多个任务执行

正确答案

Python程序通过字符串值查找枚举

如何用Python编写贝尔曼-福德算法？

如何使用Python实现计数排序算法？

Python程序用于从数组中删除给定数量的第一个项目

Python程序区分字符串的==运算符和__eq__()方法

Python程序区分字符串的==运算符和eq()方法