提问者:小点点

警告:传递了无效的参数


我在气流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/距离包/气流/型号。py:2160:PendingDeprecationWarning:传递给MySQLDoglecloudStorageOperator的参数无效。气流2.0将不再支持传递此类参数。无效参数为:*args:()**kwargs:{'provide_context':True}category=pendingdepractionwarning

提供上下文有什么问题?据我所知,使用params需要它。


共1个答案

匿名用户

provide_context不需要参数

params参数(dicttype)可以传递给任何操作员。

您将主要使用provide_contextPython算子BranchPython算子。一个很好的例子是https://airflow.readthedocs.io/en/latest/howto/operator.html#pythonoperator.

mysqltologlecloudstorageoperator没有参数provide_context,因此它被传入**kwargs,您会收到弃用警告。

如果检查PythonOperator的docstring以查找提供上下文

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

它有以下代码,如果你检查源代码:

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

因此,简单来说,它将以下带有templates\u dict的字典传递给您的函数pass inpython\u 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,
)