使用 Spark 和 Redshift 时如何优化 ETL 数据管道以实现容错？

问题描述

我正在使用 PySpark 编写一个大批量作业，该作业对 200 个表进行 ETL 处理并加载到 Amazon Redshift 中。 这 200 个表是从一个输入数据源创建的。所以只有当数据成功加载到所有 200 个表时，批处理作业才成功。批处理作业每天运行，同时将每个日期的数据附加到表中。

对于容错、可靠性和幂等性，我当前的工作流程如下：

1. 使用临时表。使用 CREATE TEMP TABLE LIKE <target_table>
创建临时 Redshift 表
2. 将数据转换并加载到临时表中。
3. 对其他 200 个表重复 1-2。
4. 开始 BEGIN 次交易。
5. 将临时表数据复制到目标表中 使用 INSERT INTO <taget_table> SELECT * FROM <staging_table>
6. END 交易
7. DROP 所有临时表。

这样我就可以保证，如果第 3 步失败（更有可能），我不必担心从原始表中删除部分数据。相反，我将简单地重新运行整个批处理作业，因为在 JDBC 断开连接后会丢弃临时表。

虽然它解决了大部分问题，但它并不优雅、笨拙并且消耗额外的时间。我想知道 Spark 和/或 Redshift 是否提供了标准工具来解决 ETL 世界中这个非常普遍的问题。

谢谢

解决方法

COPY 命令可以在事务块中。你只需要：

1. 开始
2. 将数据复制到所有表
3. 提交（如果成功）

Redshift 将为所有其他查看者维护表格的先前版本，并且他们对表格的视图在 COMMIT 之前不会更改。

您布置的进程的好处是，在事务运行期间，其他进程无法在表（ALTER TABLE 等）上获得排他锁。您的插入将比 COPY 运行得快，因此事务的打开时间将更短。如果其他进程在 ETL 运行的同时修改表，这只是一个问题，这通常不是一个好主意。