java - ETL设计: What Queue should I use instead of my SQL table and still be able to process in parallel?

Question

需要您帮助重新设计我的系统。我们有非常简单的 ETL，但也很旧，现在当我们处理大量数据时，它变得非常慢且不灵活

第一个进程是收集器进程:

收集器进程 - 始终启动

1. 收集器从队列(rabbitMQ)收集消息
2. 将消息属性(JSON 格式)解析为 java 对象(例如，如果 JSON 包含“id”、“name”和“color”等字段，我们将创建具有 int 字段“id”和字符串字段“name”的 java 对象'，和字符串字段'颜色')
3. 解析后，我们将对象作为 CSV 行写入 CSV 文件，并包含对象中的所有属性
4. 我们发送 ack 并继续处理队列中的下一条消息

处理工作流程 - 每小时发生一次

1. 名为'Loader'的进程使用SQL INFILE LOAD将所有CSV文件(收集器输出)加载到名为'Input'的数据库表，所有新行都具有'Not已处理”状态。在这个设计中，输入表就像一个队列
2. 一个名为'Processor'的进程从表中读取所有状态为“未处理”的记录，将其转换为java对象，进行一些丰富，然后将记录插入到另一个表中命名为“输出”并带有新字段，**每次迭代我们并行处理 1000 行 - 并使用 JDBC 批量更新进行数据库插入**。

此流程中的主要问题:

消息在现有流程中不灵活 - 例如，如果我想向 JSON 消息添加新属性(例如还添加 'city' )我还必须将“城市”列添加到表中(因为 CSV 文件加载)，该表包含大量数据，并且不可能在每次消息更改时添加列。

我的结论

该表不是此设计的正确选择。

我必须摆脱 CSV 写入并删除“输入”表才能拥有灵活的系统，我想也许使用队列而不是像 KAFKA 这样的表，并且可能使用 KAFKA 流等工具丰富。 - 这将使我变得灵活，每次我想向消息添加字段时，我不需要向表中添加列 我无法像今天一样并行处理这个巨大的问题。

我可以使用什么来代替表格来并行处理数据？

Answer 1

是的，使用 Kafka 会改善这一点。

摄入

当前写入 CSV 文件的进程可以发布到 Kafka 主题。这可能是 RabbitMQ 的替代品，具体取决于您的要求和范围。

加载器(可选)

以初始格式加载数据并写入数据库表的其他进程可以以您想要的格式发布到另一个 Kafka 主题。如果可以直接按照处理器想要的格式写入，则此步骤可以省略。

处理器

您使用“未处理”状态的方式是将数据视为队列的一种方式，但这是由 Kafka 中的设计处理的，它使用日志(如果关系数据库被建模为一个集合)。

处理器订阅由加载器或摄取写入的消息。它将其转换为java对象，进行一些丰富 - 但它不是将结果插入到新表中，而是可以将数据发布到新的输出主题。

不是按批处理工作:“每次迭代我们并行处理 1000 行 - 并使用 JDBC 批量更新进行数据库插入”，使用 Kafka 和流处理，这是在连续的实时流中完成的- 数据到达时。

架构可演化性

if i want for example to add new property to the json message (for example to add also 'city' ) i have to add also column 'city' to the table (because of the csv infile Load) , the table contains massive amount of data and its not possible to add column every time the message changes .

您可以使用 Avro Schema 来解决此问题发布到 Kafka 主题时。