如何加入多个Kafka主题？答案

【问题标题】：How to join multiple Kafka topics?如何加入多个Kafka主题？
【发布时间】：2018-08-19 21:55:18
【问题描述】：

所以我有...

具有通用应用程序日志 (log4j) 的第一个主题。存储诸如 HTTP API 请求/响应和警告、异常等内容……可以有多个日志与一个逻辑业务请求相关联。（这些日志在几秒钟内发生）
第二个主题包含来自上述业务请求的命令，其他服务对其采取行动。（这些命令也会在几秒钟内发生，但可能距离原始请求几分钟）
第三个主题包含由其他服务的操作生成的事件。（大多数事件在几秒钟内完成，但有些可能需要 3-5 天才能收到）

因此，单个逻辑业务请求可以通过微服务相互传递的 uuid 关联多个日志、命令和事件。

那么有哪些技术/模式可用于读取这 3 个主题并将它们全部连接为一个 json 文档，然后将它们转储到 Elasticsearch 中？

流式传输？

【问题讨论】：

【解决方案1】：

您可以使用 Kafka Streams 或 KSQL 来实现此目的。哪一个取决于您对 Java 的偏好/经验，以及您想要执行的连接的细节。

KSQL 是 Apache Kafka 的 SQL 流引擎，仅使用 SQL，您就可以针对 Kafka 主题声明流处理应用程序。您可以过滤、丰富和聚合主题。目前仅支持流表连接。你可以在this article here看到一个例子

Kafka Streams API 是 Apache Kafka 的一部分，它是一个 Java 库，可用于在 Apache Kafka 中对数据进行流处理。它实际上是建立在 KSQL 之上的，并且支持更大的处理灵活性，包括stream-stream joins。

【讨论】：

好的，所以我猜流媒体是要走的路......但我需要更多关于“如何”的信息，基于时间要求（在我的问题中解释）。
Kafka 持久化数据。只要您的保留期足以满足您的业务需求，您就可以进行您想要的联接。
好的，所以业务请求现在记录了几行，彼此间隔几秒钟，并且 4 天后从第 3 方收到了事件。所以我只是设置一个说几天的窗口吗？这是否意味着只有在最后一个事件到来时才会处理该批次？
好的。如此简单的解决方案，不需要流式传输。我可以简单地阅读每个主题并对目的地进行更新。我不需要做流来简单地加入 JSON。此外，由于消息最迟在 4 天后才收到，因此处理变得更加困难。
如果您一次只能连接 2 个流或 1 个表和 1 个流，则组合 3 个流（例如 A、B 和 C）需要 A 与 B 连接以获得 AB，然后 AB 连接用 C 得到 ABC。这对于少量数据是可行的，但是如果您需要合并更多的数据，比如 10 个或更多数据流怎么办？这仍然是这样做的方法吗？

【解决方案2】：

您可以使用 KSQL 加入流。

使用 KSQL 的好处。

缺点。

请看一看。

【讨论】：

【解决方案3】：

【讨论】：