我如何触发气流-dag使用TriggerDagRunOperator

我发现下面的链接：我如何触发气流-dag使用TriggerDagRunOperator

https://www.linkedin.com/pulse/airflow-lesson-1-triggerdagrunoperator-siddharth-anand

这的确说明了如何使用TriggerDagRunOperator来执行不同的气流DAG。文档使用了Airflow自己的示例dags，但我很难理解那些因为他们没有使用任何传感器。

有人可以解释如何使用TriggerDagRunOperator和SqlSensor开始单独的dag吗？我试图在我的SQL Server作业任务完成时启动单独的DAG。我知道如何通过使用SqlSensor来检查SQL Server作业的状态，但我不知道如何将结果附加到TriggerDagRunOperator以启动单独的DAG。

我不想使用Airflow CLI或在一个DAG中执行这两项任务。基本上，我想这只是触发dag而已。

下面是我当前的代码，这是缺少关键conditionally_trigger

# File Name: check-when-db1-sql-task-is-done 

from airflow import DAG 
from airflow.operators import TriggerDagRunOperator 
from airflow.operators import SqlSensor 
from datetime import datetime 


default_args = { 
     'owner': 'airflow', 
     'retry_delay': timedelta(minutes=5), 
} 

dag = DAG('check-when-db1-sql-task-is-done', 
     description='Check-when-DB1-SQL-task-is-done', 
     default_args=default_args, 
     schedule_interval='@once', 
     start_date=datetime.now(), 
     ) 

# returns-0-or-1-based-on-job-task-status 
sqlsensor = SqlSensor (
     task_id='sql-sensor', 
     poke_interval=30, 
     timeout=3200, 
     sql="""select last_run_outcome from msdb.dbo.sysjobsteps where job_id = '249A5A5D-6AFC-4D6B-8CB1-27C16724A450' and step_id = '1' and last_run_date = (select convert(varchar(24),getdate(),112)); """,  
     mssql_conn_id='db1', 
     dag=dag, 
     ) 

# dag-to-start 
trigger = TriggerDagRunOperator (
     task_id='start-ssh-job', 
     trigger_dag_id="qa-knime-ssh-task", 
     python_callable=conditionally_trigger, 
     params={'condition_param': True, 
       'message': 'Hello World'}, 
     dag=dag)

来源

2017-08-08 Mika Heino

我的理解是，TriggerDagRunOperator是当你想使用Python函数，以确定是否要触发SubDag。在你的代码和例子中这个函数被称为conditionally_trigger。

在你的情况下，你使用传感器来控制流量，不需要传递函数。你可以使用的，而不是SubDagOperator或TriggerDagRunOperator通过一个简单的永远真函数作为python_callable：

... 
python_callable=lambda(context, dag_run_obj):dag_run_obj, 
...

来源

2018-01-08 22:06:44 7yl4r

我如何触发气流-dag使用TriggerDagRunOperator

回答

相关问题