使用从 S3 事件调用的 AWS Lambda 触发 Airflow DAG

Question

我正在考虑构建如下工作流程：

我有一个应用程序，它将近 1000 个 csv 文件写入 s3 存储桶 MY_BUCKET 中的文件夹 MY_DIRECTORY。现在我想从 s3 存储桶解析这些文件并使用 Apache Airflow 加载到 MySQL 数据库中。

通过阅读此处的几篇文章：Airflow S3KeySensor - How to make it continue running and Airflow s3 connection using UI，我认为最好使用 AWS lambda 触发我的 Airflow DAG，一旦文件落在 s3 文件夹上，就会调用它。

作为 Airflow 和 Lambda 的新手，我不知道如何设置 lambda 来触发 Airflow DAG。在这方面，如果有人请指点一下，那将非常有帮助。谢谢。

Answer 1

创建您想要触发的 DAG，然后利用 Airflow 提供的实验性 REST API。

您可以在这里阅读它们：https://airflow.apache.org/docs/stable/api.html

特别是您想要使用以下端点：

POST /api/experimental/dags/<DAG_ID>/dag_runs

您可以在 中传递 DAG 的名称以正确触发它。此外，您可以显式传递 DAG 必须处理的文件的名称

curl -X POST \
  http://localhost:8080/api/experimental/dags/<DAG_ID>/dag_runs \
  -H 'Cache-Control: no-cache' \
  -H 'Content-Type: application/json' \
  -d '{"conf":"{\"FILE_TO_PROCESS\":\"value\"}"}'

然后使用 DAG 中的 Hook 来读取您指定的文件。