通过文件监视程序在气流中触发dag

如何解决通过文件监视程序在气流中触发dag

我使用看门狗API遍历了这篇文章，似乎正是我所需要的： https://medium.com/@phanikumaryadavilli/hacking-apache-airflow-to-trigger-dags-based-on-filesystem-events-25f822fd08c3

（代码不是我写的）

import os
import time
from airflow import DAG
from airflow.operators.dagrun_operator import TriggerDagRunoperator
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
from datetime import datetime,timedelta


dag = DAG(dag_id="test_trigger_dag_operator",default_args={"owner":"Airflow","start_date":datetime(2020,3,9)})

trigger = TriggerDagRunoperator(
    task_id="test_trigger_dag_run_operator",trigger_dag_id="dummy_operator",conf={"message": "Hello World"},dag=dag,)

class Handler(FileSystemEventHandler):
    def on_created(self,event):
        if event.event_type == 'created':
            print("file created")
            print('Executing the dag')
            trigger

def main():
    observer = Observer()
    event_handler = FileSystemEventHandler()
    observer_path = os.getcwd()
    observer.schedule(Handler(),observer_path,recursive=False)
    observer.start()

    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    
    observer.join()

if __name__ == '__main__':
    main()

不幸的是，使用作者代码，dag唯一要做的就是无条件地触发目标dag，并且main()永远不会被调用，即，也没有文件监视功能。

我对代码进行了一些小的修改，在python_callable中添加了TriggerDagRunoperator属性，并在main（context,dag_run_obj）中添加了必要的args

trigger = TriggerDagRunoperator(
    task_id="test_trigger_dag_run_operator",python_callable: main,)

并删除

if __name__ == '__main__':
    main()

部分。

现在文件监视程序正在运行，无论如何仍然会一次触发target-dag，并且调度程序在启动dag后立即挂起。 （这有点像while (true)所期望的那样）。如何才能以一种有效的方式使用所提供的代码？

解决方法

Airflow拥有自己的服务，称为DagBag Filling，可解析您的dag并将其放入DagBag中，DagBag是您在UI和元数据数据库上都看到的dag的集合。

在对文件进行DagBag填充（解析其中的DAG）时，它实际上永无止境！ 您正在DAG文件定义本身中运行该观察程序。

为避免这种情况，您需要实施Sensor

传感器-等待（轮询）特定时间的操作员，文件， 数据库行，S3键等...

，它将完全满足您的要求，但可以防止环境崩溃。 从本质上讲，这将是在自定义传感器上被覆盖的poke（）方法中重新实现您的主要功能。

您可以检查contrib仓库中的任何现有传感器，或根据需要编写自定义传感器。

如果您只想从应用程序中触发dag，则可以通过REST API

向该dagid提交POST请求

两种实现方式都可以解决您的问题