我在气流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
需要它。
provide_context
不需要参数
。
params
参数(dict
type)可以传递给任何操作员。
您将主要使用provide_context
与Python算子
,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,
)