气流过时警告传递了无效的参数

Question

我在Airflow 1.9上有以下代码：

import_op = MySqlToGoogleCloudStorageOperator(
    task_id='import',
    mysql_conn_id='oproduction',
    google_cloud_storage_conn_id='gcpm',
    provide_context=True,
    approx_max_file_size_bytes = 100000000, #100MB per file
    sql = 'import.sql',
    params={'next_to_import': NEXT_TO_IMPORT, 'table_name' : TABLE_NAME},
    bucket=GCS_BUCKET_ID,
    filename=file_name_orders,
    dag=dag)

为什么产生：

/usr/local/lib/python2.7/dist-packages/airflow/models.py:2160：PendingDeprecationWarning：无效的参数已传递给MySqlToGoogleCloudStorageOperator。在Airflow 2.0中将不再支持传递此类参数。无效的参数为：* args：（）** kwargs：{'provide_context'：True} category = PendingDeprecationWarning

是什么问题provide_context？据我所知，它的使用是必需的params。

Answer 1

provide_context不需要params。

params参数（dict类型）可以传递给任何运算符。

你会大多使用provide_context与PythonOperator，BranchPythonOperator。一个很好的例子是https://airflow.readthedocs.io/en/latest/howto/operator.html#pythonoperator。

MySqlToGoogleCloudStorageOperator没有参数provide_context，因此将其传入，**kwargs并且您会收到弃用警告。

如果你检查的文档字符串PythonOperator为provide_context：

如果设置为true，则Airflow将传递一组可以在函数中使用的关键字参数。这组kwarg与您在jinja模板中可以使用的完全对应。为此，您需要**kwargs在函数头中定义。

如果您检查源代码，它具有以下代码：

if self.provide_context:
            context.update(self.op_kwargs)
            context['templates_dict'] = self.templates_dict
            self.op_kwargs = context

因此，简单来说，它将以下字典与templates_dict传递给您的函数一起传递python_callable：

{
    'END_DATE': ds,
    'conf': configuration,
    'dag': task.dag,
    'dag_run': dag_run,
    'ds': ds,
    'ds_nodash': ds_nodash,
    'end_date': ds,
    'execution_date': self.execution_date,
    'latest_date': ds,
    'macros': macros,
    'params': params,
    'run_id': run_id,
    'tables': tables,
    'task': task,
    'task_instance': self,
    'task_instance_key_str': ti_key_str,
    'test_mode': self.test_mode,
    'ti': self,
    'tomorrow_ds': tomorrow_ds,
    'tomorrow_ds_nodash': tomorrow_ds_nodash,
    'ts': ts,
    'ts_nodash': ts_nodash,
    'yesterday_ds': yesterday_ds,
    'yesterday_ds_nodash': yesterday_ds_nodash,
}

因此，可以在函数中使用它，如下所示：

def print_context(ds, **kwargs):
    pprint(kwargs)
    ti = context['task_instance']
    exec_date = context['execution_date']
    print(ds)
    return 'Whatever you return gets printed in the logs'


run_this = PythonOperator(
    task_id='print_the_context',
    provide_context=True,
    python_callable=print_context,
    dag=dag,
)