文件上传时触发Airflow任务

Question

我在想一个办法，就是当用户往DAG目录上传任何文件时，如何触发Airflow的任务。请问有什么Airflow的操作符可以帮忙吗？

我知道有一个叫FileSensor的操作符，但我理解它是需要特定的文件上传到目录里。

如果Airflow没有这样的操作符，那我该怎么做呢？

Answer 1

你可以创建一个用来存放上传文件的文件夹，比如叫做 'user_upload/'。接下来，你可以通过 os.path.getmtime 或 os.path.getctime 来获取这个文件夹下所有文件的修改时间或创建时间。然后，你可以把这些时间和一些模板变量，比如 data_interval_start 或 ts 进行比较，从中找出新的文件。如果没有新的文件，你就可以使用 AirflowSkipException 跳过后续的任务。