如何覆盖来自 Airflow 的 python 脚本中的参数值？

如何解决如何覆盖来自 Airflow 的 python 脚本中的参数值？

我有 python 脚本 aws_test.py，我通过气流调用 EScoperator 我在 .py 脚本中有名为 instance 的参数，我想使用 airlfow dag 覆盖它。如何做到这一点.

要从 DAG 传递到要覆盖的 .py 脚本的值

实例='ins12'

我当前的 Airflow DAG 代码

dag = DAG(
    'execute_sample_job',default_args=default_args,schedule_interval='@daily',max_active_runs=1
)

set_input_task = Pythonoperator(
    task_id='set_params',python_callable=set_input,provide_context=True,dag=dag)

execute = ECSOperator(
    task_id='execute',launch_type='FARGATE',task_deFinition=Helpers.get_param('task_def'),cluster="ds-notebooks",awslogs_group=Helpers.get_param('awslogs_group'),awslogs_stream_prefix=Helpers.get_param('awslogs_stream_prefix'),network_configuration={
        'awsvpcConfiguration': {
            'securityGroups': [Helpers.get_param('security_group')],'subnets': [Helpers.get_param('subnet1'),Helpers.get_param('subnet2'),Helpers.get_param('subnet3')]
        }
    },overrides={
        'containerOverrides': [
            {
                'name': 'notebook','environment': [
                    {
                        'name': 'INPUT','value': 'tasks/python/aws_test.py'
                    },{
                        'name': 'ParaMETERS','value': json.dumps(
                            dict(
                                instance='ins12',)
                        )
                    },],},dag=dag)

execute.set_upstream(set_input_task)

解决方法

在 DagRun conf 中传递实例名称。例如：

{"instance_name": "fooName"}

然后您可以在 overrides 中添加 reference it，因为它是一个模板字段：

...
'value': '{"instance": "{{ dag_run.conf["instance"] }}"}'
...

编辑

或者，如果您在 set_input_task 中生成实例名称，请执行以下操作：

    overrides={
        'containerOverrides': [
            {
                'name': 'notebook','environment': [
                    {
                        'name': 'INPUT','value': 'tasks/python/aws_test.py'
                    },{
                        'name': 'PARAMETERS','value': '{"instance": "{{ task_instance.xcom_pull["set_input_task"] }}"}'
                    },],},

通过这种方式，您可以使用 XCom 向它们传递名称。这假设您的 set_input 函数返回一个字符串作为实例名称。