气流：从上游任务访问模板字段

Question

我有两个任务，一个是自定义运算符，它有一个模板字段 (snapshot_date_str)，它将在“xcom”中设置字段，另一个运算符是 S3Sensor，bucket_key 需要在第一个任务中设置的模板字段。

Dag 定义：

SNAPSHOT_DATE = datetime.now().date()
S3_BUCKET = 'test-s3'
TENANT = 'test'

dag = DAG('template_fields_dag',
          default_args=default_args,
          schedule_interval='@hourly',
          concurrency=1,
          catchup=False)

t1 = ContextInitOperator(task_id='set_context', snapshot_date=SNAPSHOT_DATE, tenant=TENANT, dag=dag)

file_task = S3KeySensor(task_id="s3_file_sensor",
                        aws_conn_id='s3_connection',
                        bucket_key='test/{{ snapshot_date_str }}/abc.csv',
                        bucket_name=S3_BUCKET,
                        wildcard_match=True,
                        poke_interval=10,
                        timeout=60,
                        dag=dag)
t1 >> file_task

而我的自定义ContextInitOperator在xcom中设置了模板字段snapshot_date_str。

class ContextInitOperator(BaseOperator):

    template_fields = ('snapshot_date_str',)

    @apply_defaults
    def __init__(
            self,
            snapshot_date,
            *args, **kwargs):
        super(ContextInitOperator, self).__init__(*args, **kwargs)
        self.snapshot_date_str = snapshot_date.strftime('%Y-%m-%d')

    def execute(self, context):
        context['task_instance'].xcom_push(key='snapshot_date_str', value=self.snapshot_date_str)

bucket_key 需要路径中的snapshot_date_str。

我对 Python 和 Airflow 还不是很满意，是不是缺少一些基本的东西？任何帮助将不胜感激。

Answer 1

来自documentation，您可能需要在以下行中做一些事情

bucket_key="test/{{ task_instance.xcom_pull(task_ids='set_context', key='snapshot_date_str') }}/abc.csv"

Answer 2

如果这只是为了获取格式化的日期，Airflow 可以为您提供帮助。根据您的需要，您可以使用以下预定义变量：

bucket_key='test/{{ ds }}/abc.csv',

今天，

bucket_key='test/{{ yesterday_ds }}/abc.csv',

昨天和

bucket_key='test/{{ tomorrow_ds }}/abc.csv',

明天。在此处查看所有可用的宏：https://airflow.apache.org/code.html#macros

这意味着您的ContextInitOperator 可能会被删除。

bucket_key 也是一个模板字段，可以在源文件 (https://airflow.incubator.apache.org/_modules/airflow/operators/sensors.html) 中看到，因此可以使用 Jinja 变量。

Airflow 中的日期处理方式略有不同，因此您可能需要进行试验以获得所需的结果（来自 https://airflow.incubator.apache.org/scheduler.html）：

请注意，如果您在一天的 schedule_interval 上运行 DAG，则标记为 2016-01-01 的运行将在 2016-01-01T23:59 之后不久触发。换言之，作业实例会在其涵盖的时间段结束后启动。