气流将参数传递给相关任务

Question

Airflow中的依赖任务传递参数的方法是什么？我有很多 bashes 文件，我正在尝试将这种方法迁移到气流，但我不知道如何在任务之间传递一些属性。

这是一个真实的例子：

#sqoop bash template
sqoop_template = """
        sqoop job --exec {{params.job}} -- --target-dir {{params.dir}} --outdir /src/
    """

s3_template = """
        s3-dist-cp --src= {{params.dir}} --dest={{params.s3}}
    """



#Task of extraction in EMR
t1 = BashOperator(
        task_id='extract_account', 
        bash_command=sqoop_template, 
        params={'job': 'job', 'dir': 'hdfs:///account/' + time.now().strftime("%Y-%m-%d-%H-%M-%S")},
        dag=dag)
#Task to upload in s3 backup.
t2 = BashOperator(
        task_id='s3_upload',
        bash_command=s3_template,
        params={}, #here i need the dir name created in t1
        depends_on_past=True
    )

t2.set_upstream(t1)

在 t2 中，我需要访问在 t1 中创建的目录名称。

解决方案

#Execute a valid job sqoop
def sqoop_import(table_name, job_name):
    s3, hdfs = dirpath(table_name)
    sqoop_job = job_default_config(job_name, hdfs)
    #call(sqoop_job)
    return {'hdfs_dir': hdfs, 's3_dir': s3}

def s3_upload(**context):
    hdfs = context['task_instance'].xcom_pull(task_ids='sqoop_import')['hdfs_dir']
    s3 = context['task_instance'].xcom_pull(task_ids='sqoop_import')['s3_dir']
    s3_cpdist_job = ["s3-dist-cp", "--src=%s" % (hdfs), "--dest=%s" % (s3)]
    #call(s3_cpdist_job)
    return {'s3_dir': s3} #context['task_instance'].xcom_pull(task_ids='sqoop_import')

def sns_notify(**context):
    s3 = context['task_instance'].xcom_pull(task_ids='distcp_s3')['s3_dir']
    client = boto3.client('sns')
    arn = 'arn:aws:sns:us-east-1:744617668409:pipeline-notification-stg'
    response = client.publish(TargetArn=arn, Message=s3)
    return response

这不是最终的解决方案，因此欢迎进行改进。谢谢

Answer 1

查看 XComs - http://airflow.incubator.apache.org/concepts.html#xcoms。这些用于任务之间的状态通信。

Answer 2

我认为 Airflow 不是用来管理状态的。您应该为任务使用数据库来交换状态。

气流将参数传递给相关任务

Airflow pass parameters to dependent task

bash

airflow

解决方案