气流：创建一次性动态子对象

Question

我正在创建一个dag，它执行在某些数据库中预定义的任务。 任务完成后，我会更新它们的执行时间，直到它们应该再次执行为止。每个任务的目的基本上都是进行sql单元测试。在

到目前为止我想做的是

1. 创建父主dag
2. 从数据库中获取任务列表
3. 对于每一行（任务）-我正在创建包含执行过程的subdag
4. 当所有子任务完成后-我正在更新任务的渗出时间

目前它在第一次运行后失败。显示的错误Broken DAG: [/usr/local/airflow/src/dags/d06-query_validations/d06-query_validations_daily.py] list index out of range。 拜托，帮我弄清楚问题出在哪里

我目前所做的努力：

default_args = {
'owner': 'airflow',
'depends_on_past': False,
'start_date': datetime(2018, 11, 25, 8, 15),
'wait_for_downstream': True,
'email': email_list,
'email_on_failure': True,
'email_on_retry': False
}


def getValidationsToRun():
    start_time = datetime.now()
    conn = MySqlHook(mysql_conn_id='mysql_main', kwargs={"charset": "utf8"})
    query = ReadTextFile('/d06-query_validations/get_validations.sql')
    logging.log(logging.INFO, "Extract Query={}".format(query))
    records = conn.get_pandas_df(query)
    logging.log(logging.INFO, "Extract completed. it took: 
    {}".format(str(datetime.now() - start_time)))
    return records


def create_subdag(parent_dag_name, child_dag_name, validation):
    inner_dag = DAG(
        %s.%s' % (parent_dag_name, child_dag_name),
        default_args=default_args.copy(),
        schedule_interval='@once'
    )
    QueryValidationFlow(
        dag=inner_dag,
        validation_name=validation.validationName,
        title=validation.messageTemplate,
        query=validation.query,
        expected_result=validation.expectedResult,
        source_db=validation.source,
        emails=validation.emailRecipients.split(',')
        )
        return inner_dag


def create_subdag_operator(parent_dag, validation):
    child_dag_name = 'subdag_{}'.format(validation.validationName)
    parent_dag_name = parent_dag.dag_id
    subdag = SubDagOperator(
        task_id=child_dag_name,
        dag=parent_dag,
        subdag=create_subdag(parent_dag_name, child_dag_name, validation)
    )
    return subdag


def create_subdag_operators(parent_dag, validations):
    subdag_list = [create_subdag_operator(parent_dag, row) for index, row in validations.iterrows()]
    # chain subdag operators together
    helpers.chain(*subdag_list)
    return subdag_list


# (top-level) DAG & operators
dag = DAG(dag_id='d06-query_validations', schedule_interval='0 * * * *', 
default_args=default_args, catchup=False)

curr_validations = getValidationsToRun()

curr_validation_ids = ",".join(["'%s'" % str(validationId) for validationId in curr_validations["validationId"]])

dummy_op_start = DummyOperator(task_id='d06-op_start', dag=dag)

subdag_ops = create_subdag_operators(dag, curr_validations)

update_execution_time = MySqlOperator(
    task_id='d06-update_execution_time',
    sql=ReadTextFile('/d06- 
    query_validations/update_validations.sql').format(curr_validation_ids),
    mysql_conn_id='mysql_main',
    retries=5,
    execution_timeout=timedelta(minutes=2),
    retry_delay=60,
    dag=dag
)

dummy_op_start >> subdag_ops[0]
subdag_ops[-1] >> update_execution_time

Answer 1

仅供参考，您的DAG文件的直接上下文中的所有内容都由flow web服务器和airflow调度器在一个循环中执行，以确定DAG中的内容。即使在DAG文件夹中不生成DAG的python文件也会发生这种情况。对于DAG没有计划或在UI或DB中已禁用的DAG文件，也会发生这种情况。它这样做是因为任何python文件都可能动态地生成一个新的DAG。在

所以这是运行了很多次：

def getValidationsToRun():
  start_time = datetime.now()
  conn = MySqlHook(mysql_conn_id='mysql_main', kwargs={"charset": "utf8"})
  query = ReadTextFile('/d06-query_validations/get_validations.sql')
  logging.log(logging.INFO, "Extract Query={}".format(query))
  records = conn.get_pandas_df(query)
  logging.log(logging.INFO, "Extract completed. it took: 
  {}".format(str(datetime.now() - start_time)))
  return records

我敢肯定你会看到如果你检查你的调度日志。在

我怀疑有时结果是空的，因此subdag_ops[0]超出范围。在

还有

^{pr2}$

指示您尚未阅读有关使用模板化字段和参数的信息。它应该更像：

sql='./d06- 
query_validations/update_validations.sql',
params={'val_ids': curr_validation_ids},

其中包含{{ params.val_ids }}的sql文件。在

也许Astronomer docs for templating比Airflow ones更有帮助？在