根据表更新何时启动DAG

Question

我有一个10-15步骤的DAG流程，当天更新了一个BigQuery表之后，我需要启动。可以使用简单的选择SQL语句检查表：

select * from sample_table where update_date = current_date()

当查询返回值时，我们需要在气流/云作曲家中启动一系列任务。

Answer 1

使用传感器节点创建另一个DAG，该传感器节点检查BigQuery表格以获取当天的更新。

在更新表的地方，请使用TriggerDagrunoperator触发您的DAG。

如果您的大查询表的更新是不可预测的，则可以按计划运行此主管DAG

from typing import Optional, Union, Sequence

from airflow.operators.trigger_dagrun import TriggerDagRunOperator
from airflow.providers.google.cloud.operators.bigquery import (
    BigQueryValueCheckOperator
)
from airflow.utils.context import Context

class BigQueryTableSensor(BaseSensorOperator, BigQueryCheckOperator):
    def __init__(
        self,
        *,
        sql: str,
        gcp_conn_id: str = 'google_cloud_default',
        use_legacy_sql: bool = True,
        location: Optional[str] = None,
        impersonation_chain: Optional[Union[str, Sequence[str]]] = None,
        labels: Optional[dict] = None,
    ):
        super().__init__(**kwargs) # initialize base sensor
        BigQueryCheckOperator.__init__(
            self,
            sql,
            gcp_conn_id,
            use_legacy_sql,
            location,
            impersonation_chain,
            labels,
        )
         
    def poke(self, context: 'Context') -> bool:
        return BigQueryCheckOperator.execute(self, context)


# dag initialisation...
with DAG(
    dag_id="dag_process_supervisor",
    start_date=pendulum.datetime(2022, 05, 25, tz="UTC"),
    catchup=True,
    schedule_interval="@hourly",
) as dag:
    wait_for_table_update = BigQueryTableSensor(
       sql = '''
          SELECT COUNT(*) > 1
             FROM sample_table 
          WHERE update_date = current_date();
       ''',
       pass_value=True,
       poke_interval=60 * 15, # 15 mins
       mode='poke'
    )
    trigger_dag_process = TriggerDagRunOperator(
        task_id="trigger_dag_process",
        trigger_dag_id="dag_process",
    )
    wait_for_table_update >> trigger_dag_process