python - PythonOperator 的运行时配置

Question

我找不到任何有关如何使用通过“-c”传递的 JSON 变量的工作文档，例如回填工作。

我一直在打印我的 python 任务 **kwargs 来找出答案，但我仍然无法确定它。 Provide_context=True

有人能指出我正确的方向吗？

所以，我想做的是:

airflow backfill mydag -c '{"override":"yes"}' -s 2018-12-01 -e 2018-12-12

我有一个PythonOperator:

PythonOperator(
    task_id = 'task_identifier',
    python_callable = 'run_task',
    op_kwargs = {
        'task_conf': task_config 
    },
    provide_context=True,
    dag = this_dag
)

在 run_task 中，我想访问覆盖变量:

def run_task(*args, **kwargs): 

    dag_run = kwargs.get('dag_run')
    logging.info(kwargs['dag_run'].conf.get('override'))

但我找不到访问此覆盖变量的方法

[2018-12-17 10:07:24,649] {models.py:1760} ERROR - 'NoneType' object has no attribute 'get'
Traceback (most recent call last):
  File "/home/daimonie/.local/lib/python3.6/site-packages/airflow/models.py", line 1659, in _run_raw_task
    result = task_copy.execute(context=context)
  File "/home/daimonie/.local/lib/python3.6/site-packages/airflow/operators/python_operator.py", line 95, in execute
    return_value = self.execute_callable()
  File "/home/daimonie/.local/lib/python3.6/site-packages/airflow/operators/python_operator.py", line 100, in execute_callable
    return self.python_callable(*self.op_args, **self.op_kwargs)
  File "/home/daimonie/airflow/dags/dag_scheduled_queries.py", line 65, in run_query
    logging.info(kwargs['dag_run'].conf.get('override'))

编辑:我确实找到了一个配置设置和描述，似乎表明这些参数需要已经在代码中设置

# Whether to override params with dag_run.conf. If you pass some key-value pairs through `airflow backfill -c` or
# `airflow trigger_dag -c`, the key-value pairs will override the existing ones in params.
dag_run_conf_overrides_params=True

donot_pickle 参数已设置为 False。

Answer 1

对于任何想知道错误原因的人

AttributeError: 'NoneType' object has no attribute 'get

当尝试从 dag_run 的 conf 属性检索键/值时，是因为 conf 参数传递给 airflow backfill mydag -c需要配置为:

'{"conf": {"key1": "value", ...}}'

不是

'{"key1": "value", ...}'

然后您可以从上下文中引用它，例如

def do_something(**context):
    value = context['dag_run'].conf['key']
    ...