amazon-s3 - 使用从 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和 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 来读取您指定的文件。