Apache Airflow：关于动态任务和并行性的问题
Apache Airflow: Question about Dynamic Tasks and Parallelism
我被要求编写更多 "professional" 版本的 ETL 脚本。我所说的专业是指：监督（日志记录和电子邮件警报）、并行性和调度。带着这些目标，我开始阅读 Apache Airflow 文档。
一切看起来都很棒，但我不知道并行 的最佳方法是什么。原始脚本首先下载对象 ID 列表。然后，对于每个 id，它会查询另一个端点，进行一些基本计算（生成新字段）并将数据存储到 CSV 文件中。
并行执行此操作的最佳方法是什么？为每个对象动态启动任务并稍后加入数据？使用 XCOM 传输数据是个好主意吗？
如果有数千个对象怎么办？另一件很棒的事情是，如果查询失败则重试查询。
我知道在这种情况下我必须使用 celery executor 并生成一些 worker 来处理所有任务。
Apache Airflow 是合适的工具吗？
谢谢！
你绝对可以在 Airflow 中做到这一点。我有类似的情况，我的 DAG 首先检查有多少对象，然后使用基本的 for 循环为每个对象创建任务。操作员和任务被唯一命名并分配给每个对象。然后并行级别可以通过基本的 Airflow 配置选项来控制。为了在最后将所有东西连接在一起，我会对 xcoms 保持警惕，因为它们会占用大量的 db space。您可能会考虑暂时存储平面文件，然后合并它们以创建最终输出。
我被要求编写更多 "professional" 版本的 ETL 脚本。我所说的专业是指：监督（日志记录和电子邮件警报）、并行性和调度。带着这些目标，我开始阅读 Apache Airflow 文档。
一切看起来都很棒，但我不知道并行 的最佳方法是什么。原始脚本首先下载对象 ID 列表。然后，对于每个 id，它会查询另一个端点，进行一些基本计算（生成新字段）并将数据存储到 CSV 文件中。
并行执行此操作的最佳方法是什么？为每个对象动态启动任务并稍后加入数据？使用 XCOM 传输数据是个好主意吗？ 如果有数千个对象怎么办？另一件很棒的事情是，如果查询失败则重试查询。
我知道在这种情况下我必须使用 celery executor 并生成一些 worker 来处理所有任务。
Apache Airflow 是合适的工具吗？ 谢谢！
你绝对可以在 Airflow 中做到这一点。我有类似的情况，我的 DAG 首先检查有多少对象，然后使用基本的 for 循环为每个对象创建任务。操作员和任务被唯一命名并分配给每个对象。然后并行级别可以通过基本的 Airflow 配置选项来控制。为了在最后将所有东西连接在一起，我会对 xcoms 保持警惕，因为它们会占用大量的 db space。您可能会考虑暂时存储平面文件，然后合并它们以创建最终输出。