任务实例未定义答案

【问题标题】：Task Instance is not defined任务实例未定义
【发布时间】：2020-12-11 22:46:12
【问题描述】：

我有一个传递给 xcom 的对象，我想从操作员那里读取。

这是我的操作员：

load_csv = GCSToBigQueryOperator(
    task_id='gcs_to_bigquery',
    bucket='test',
    source_objects=['{{ execution_date.strftime("%Y-%m") }}'],
    providers=True,
    destination_project_dataset_table=f'{stg_dataset_name}' + '.' + '{{ execution_date.strftime("%Y_%m") }}',
    schema_fields={{ti.xcom_pull(task_ids='print_the_context')}},
    write_disposition='WRITE_TRUNCATE',
    provide_context=True,
    dag=dag)

我想将 xcom 中的值传递给 schema_fields 变量。

我正在尝试使用以下模板访问对象 {{ti.xcom_pull(task_ids='print_the_context')}} 但我有 it is not defined...

这里有什么问题？

【问题讨论】：

标签： airflow

【解决方案1】：

不幸的是，目前这是不可能的

为了能够在operator 参数中利用macros，相应的字段必须在运算符的源代码中定义为template_fields

但是在GCSToBigQueryOperator的源码这里，我看到schema_fields is missing from template_fields

template_fields = ('bucket', 'source_objects',
                    'schema_object', 'destination_project_dataset_table')

因此，您无法通过 XCOM 模板为 schema_fields 提供值

也就是说，虽然我不了解 GCSToBigQueryOperator 的内部结构，但我可以看到 2 个可能的解决方案

（直截了当）改用schema_object field

:param schema_object: If set, a GCS object path pointing to a .json file that
     contains the schema for the table. (templated)
     Parameter must be defined if 'schema_fields' is null and autodetect is False.
 :type schema_object: str

您可以尝试将其子类化并在template_fields 中包含schema_fields

有趣的阅读

【讨论】：