如何在 BigQueryOperator 上参数化 write_disposition？

Question

我正在尝试将 jenkins 上的当前管道调整为由气流编排。 我们的需求之一是能够通过手动触发器重新处理日期循环中的历史数据。

假设我刚刚在我的管道/dag 上添加了一个新阶段，并且我想在过去 n 天内执行它，而不更改已创建的表，我可以将所有 BigQueryOperators 设置为在 write_disposition 上工作：'WRITE_EMPTY'

我知道如果 write_disposition 是一个模板字段，那就这么简单：

BigQueryOperator(
            task_id="table_x",
            bql='sql_folder/my_query.sql',
            destination_dataset_table="project_id.dataset_id.my_table_{{ dag_run.conf["date_suffix"] }}",
            write_disposition='{{ dag_run.conf["write_disposition"] }}',
            params={
                 'event_date': "{{ dag_run.conf["event_date"] }}"
            }
        )

在日期循环上调用 dag：

airflow trigger_dag 'example_dag_conf' -r 'run_id' --conf '{"date_suffix":"20191213", "event_date":"2019-12-13", "write_disposition": ""}'

但事实并非如此。

有什么解决方法吗？还是另一种方法？

Answer 1

您可以为您的用例创建一个继承自 BigQueryOperator 的 CustomBigQueryOperator。

例子：

class CustomBigQueryOperator(BigQueryOperator):
    template_fields = ('sql', 'destination_dataset_table', 'labels', 'write_disposition')

并在你的 DAG 中使用它。