mysql - 通过Airflow更新SQL数据库中的大量数据

Question

我在 CloudSQL 中有一个很大的表，需要每小时更新一次，我正在考虑将 Airflow 作为一个潜在的解决方案。从 Airflow 更新 CloudSQL 数据库中的大量数据的最佳方法是什么？

约束是:

1. 作业运行时表格仍需可读

2. 该表需要可写，以防其中一个作业超时并且 2 个作业最终同时运行

我的一些想法:

1. 加载数据需要更新到pandas框架并运行pd.to_sql

2. 将数据加载到 Cloud Storage 中的 csv 中并执行 LOAD DATA LOCAL INFILE

3. 将数据加载到内存中，将其分成 block ，然后运行一个多线程进程，每个进程使用共享连接池逐 block 更新表，以防止耗尽连接限制

Answer 1

我最近的airflow相关ETL项目可以给你一个引用。

• 输入DB:LargeDB(十亿行级Oracle)
• Interim DB:Mediam DB(千万级HD5文件)
• 输出DB:Mediam DB(千万级mysql)

据我所知，写入数据库是此类 ETL 过程的主要 block 。如您所见，

• 对于中间阶段，我使用 HD5 作为中间 DB 或文件进行数据转换。 pandas to_hdf 函数为大数据提供秒级性能。就我而言，20 百万行写入 hdf5 的时间不到 3 分钟。以下是 pandas IO 的性能基准测试。 HDF5 格式是前三名最快且最流行的格式。 https://pandas.pydata.org/pandas-docs/stable/user_guide/io.html#io-perf

• 对于输出阶段，我使用带有 chunk_size 参数的 to_sql。为了加快 to_sql 的速度，您必须手动将列类型映射到数据库列类型和长度，特别是字符串或 varchar 格式。通过手动映射，to_sql 将映射为 blob 格式或 varchar(1000)。默认模式比手动映射模式慢10倍。通过to_sql(chunksize模式)总共2000万行写入数据库大约需要20分钟。

如果您喜欢这个答案，请投票