python - 如何在一项任务中运行 dag

Question

default_args = {
"owner": "airflow_admin",
"start_date": datetime(2017, 8, 29),
"schedule_interval":'*/5 * * * *'
 }
with DAG(
'my_first_dag',
catchup=False,
default_args=default_args,
) as dag:

   task1 = BashOperator(
   task_id='id1',
   bash_command=bash_.format('db1', 'tab1'))

task1

有谁知道如何通过一项任务启动 dag，上面的代码不起作用，dag 已正确启动

Answer 1

确保您的缩进正确。您的任务应在 DAG 上下文管理器的范围内定义，否则您必须向您的任务提供 dag 参数:

尝试:

default_args = {
    "owner": "airflow_admin",
    "start_date": datetime(2017, 8, 29),
    "schedule_interval": "0 5 * * *"
}
with DAG('my_first_dag', catchup=False, default_args=default_args) as dag:
    task1 = BashOperator(
        task_id='id1',
        bash_command=bash_.format('db1', 'tab1'))

或者不使用上下文管理器:

default_args = {
    "owner": "airflow_admin",
    "start_date": datetime(2017, 8, 29),
    "schedule_interval": "0 5 * * *"
}
dag = DAG('my_first_dag', catchup=False, default_args=default_args)
task1 = BashOperator(
        task_id='id1',
        bash_command=bash_.format('db1', 'tab1'),
        dag=dag)