2.0中的Airflow dag和任务装饰器:如何将配置参数传递给任务？

default_args = {
    'param1': 'value1',
    'param2': 'value2'
}

with DAG('my_dag', default_args=default_args, schedule_interval='@daily') as dag:
    ...

@task
def my_task(**kwargs):
    param1 = kwargs['dag_run'].conf.get('param1')
    param2 = kwargs['dag_run'].conf.get('param2')
    ...

dag_run = DagRun.find(dag_id='my_dag_id', execution_date='2022-01-01')[0]
conf = {'param1': 'new_value1', 'param2': 'new_value2'}
dag_run.conf = conf
dag_run.run()