airflow - 使用 Airflow 在数据库之间移动和转换数据

Question

使用 Airflow，我从 MySQL 数据库中提取数据，使用 Python 对其进行转换，然后将其加载到 Redshift 集群中。

目前我使用 3 个 Airflow 任务:它们通过在本地磁盘上写入 CSV 来传递数据。

如何在不写入磁盘的情况下执行此操作？我应该用 python 编写一项大任务吗？ (这会降低能见度)

编辑:这是一个关于 Airflow 的问题，以及选择任务粒度以及如何在任务之间传递数据的最佳实践。这不是一个关于数据迁移或 ETL 的一般问题。在本问题中，ETL 仅用作 Airflow 任务的工作负载示例。

Answer 1

您可以通过多种方式实现这一目标:

如果您使用适用于 MySQL 的 AWS RDS 服务，则可以使用 AWS Data Pipeline 将数据从 MySQL 传输到 Redshift。他们在 AWS Data Pipeline 中有内置模板来执行此操作。您甚至可以安排从 MySQL 到 Redshift 的增量数据传输

http://docs.aws.amazon.com/datapipeline/latest/DeveloperGuide/dp-template-redshift.html

你的 table 有多大？如果你的表不是太大，你可以使用 Pandas DataFrame 或元组将整个表读入 python，然后将其传输到 Redshift。即使您有很大的表，您仍然可以分块读取该表并将每个 block 推送到 Redshift。如果将表读入 Pandas 中，则在内存使用方面效率很低。

在 Airflow 中创建不同的任务不会有太大帮助。您可以创建一个函数并使用 PythonOperator 在 dag 中调用该函数，或者创建一个 python 脚本并在 dag 中使用 BashOperator 执行它