之前我们简单总结过串联任务的 DAG 脚本如何编写，具体可以参考一下 串联任务。下面我们改写一下脚本，来让脚本并行跑任务。将任务处理成并行 task，最大的好处就是可以极大地节省时间，下面直接贴一下并联代码：

    1. def run_case_sql(i=0):
    2.     """运行sql函数"""
    3.     batch_date = airflow_get_date(today_key, 0)
    4.     context = {"BATCH_DATE": batch_date}
    5.     sql_file = os.path.join(sql_path, "case.sql")
    6.     runner = ImpalaHiveSQLRunner(db_name="impala", config=databases, filename=sql_file, context=context)
    7.     runner.run_sql_block(sql_name="sql" + str(i))
    8.     runner.close()
    11. def run_phone_sql(i=0):
    12.     """运行sql函数"""
    13.     batch_date = airflow_get_date(today_key, 0)
    14.     context = {"BATCH_DATE": batch_date}
    15.     sql_file = os.path.join(sql_path, "phone.sql")
    16.     runner = ImpalaHiveSQLRunner(db_name="impala", config=databases, filename=sql_file, context=context)
    17.     runner.run_sql_block(sql_name="sql" + str(i))
    18.     runner.close()
    20. default_args = {
    21.     "owner": "yumingmin",
    22.     "depends_on_past": False,
    23.     "start_date": days_ago(1),
    24.     "email": ["yu_mingm623@163.com"],
    25.     "email_on_failure": False,
    26.     "email_on_retry": False,
    27.     "retries": 60,
    28.     "retry_delay": timedelta(minutes=5),
    29.     "catchup": False
    30. }
    32. dag = DAG(
    33.     project_name,
    34.     default_args=default_args,
    35.     description=project_name,
    36.     schedule_interval='30 8 * * *')
    38. init_op = PythonOperator(
    39.     task_id="init",
    40.     python_callable=init,
    41.     dag=dag)
    43. for j in range(0, 9):
    44.     parallel_op = PythonOperator(
    45.         task_id="run_case_sql" + str(j + 1),
    46.         pyttaskflowshon_callable=run_case_sql,
    47.         op_kwargs={"i": j + 1},
    48.         dag=dag)
    49.     init_variable_op >> parallel_op
    51. for j in range(0, 2):
    52.     parallel_op = PythonOperator(
    53.         task_id="run_phone_sql" + str(j + 1),
    54.         python_callable=run_case_sql,
    55.         op_kwargs={"i": j + 1},
    56.         dag=dag)
    57.     init_variable_op >> parallel_op

    💡 注意：如果是并行任务，一定是要有一个并行的分支点才可以。