代码之家 › 专栏 › 技术社区 › vdolez

在气流中将返回值从操作员传递给后续操作员

google-cloud-composer airflow python

0

vdolez · 技术社区 · 8 年前

我想给 source_objects 用于 GoogleCloudStorageToBigQueryOperator

字符串索引必须是整数,而不是unicode

我怎么得到 return 价值 get_file_name 在DAG示波器上用XCOM?
xcom_pull 在DAG作用域中的函数而不必提供上下文?在我看来,任务实例不需要提供上下文。

我想到的事情:

重写运算符并将XCOM作为参数

我想做的是:

而且,似乎有些运算符字段使用了一个名为 templated_field ,模板字段后面的机制是什么?不仅仅是为了 PythonOperator BashOperator ?

蟒蛇 不返回 TaskInstance

with DAG('bq_load_file_from_cloud_function', default_args=default_args) as dag:

    def get_file_name_from_conf(ds, **kwargs):
        fileName = kwargs['dag_run'].conf['fileName']
        return [fileName]

    get_file_name = PythonOperator(
        task_id='get_file_name',
        provide_context=True,
        python_callable=get_file_name_from_conf)

    # t1, t2 and t3 are examples of tasks created by instantiating operators
    bq_load = GoogleCloudStorageToBigQueryOperator(
        task_id='bq_load', 
        bucket='src_bucket', 
        #source_objects=['data.csv'], 
        source_objects=get_file_name.xcom_pull(context='', task_ids='get_file_name'), 
        destination_project_dataset_table='project:dataset.table', 
        write_disposition='WRITE_EMPTY')

    bq_load.set_upstream(get_file_name)

我对Python和airlow有点陌生。我想这类事情应该是微不足道的。我肯定这里有我误会的地方。

1 回复 | 直到 8 年前

1

2

vdolez 8 年前

经过多次测试,我想出了这个解决方案,感谢tobi6的评论,他给了我正确的方向。我不得不使用 模板字段 特色。

以下是最终代码:

with DAG('bq_load_file_from_cloud_function', default_args=default_args) as dag:

    def get_file_name_from_conf(ds, **kwargs):
        return kwargs['dag_run'].conf['fileName']

    get_file_name = PythonOperator(
        task_id='get_file_name',
        provide_context=True,
        python_callable=get_file_name_from_conf)

    bq_load = GoogleCloudStorageToBigQueryOperator(
        task_id='bq_load', 
        bucket='src_bucket', 
        source_objects=["{{ task_instance.xcom_pull(task_ids='get_file_name') }}"],
        destination_project_dataset_table='project:dataset.table', 
        write_disposition='WRITE_APPEND')

    bq_load.set_upstream(get_file_name)