来自Timestamp的两个不同KAFKA主题的Spark聚集事件

Question

假设一个具有以下两个主题的Kafka系统：

• 创建的
• 删除，

它们用于宣传项目的创建和删除。

Kafka中事件的结构是JSON，这两个主题都是相同的：

{
  "id" : "a1cf621a-2a96-4b70-9dd6-3c54a2819eef"
  "timestamp": "2022-01-05T07:31:04.913000"
}

现在，Spark（Scala）可以如何积累删除和创建的量，以使我们通过时间戳获得许多当前项目。

假设Kafka主题中的以下事件

：创建

{"id":"1","timestamp":"2022-01-01T00:00:00.000000"}
{"id":"2","timestamp":"2022-01-02T00:00:00.000000"}

主题：已删除

{"id":"2","timestamp":"2022-01-03T00:00:00.000000"}
{"id":"1","timestamp":"2022-01-04T00:00:00.000000"}

，这基本上是指：

• 2022-01-01：1创建了项目，总计项目计数为1
• 2022-- 01-02：1创建项目，项目的总数为2
• 2022-01-03：1项目已删除，总计项目计数为1
• 2022-01-04：1项目已删除，总计计数为0，是

0该程序的结果应该是每个时间戳的项目计数，例如：在这里：

----------------------------------------
| timestamp                    | count |
----------------------------------------
| 2022-01-01T00:00:00.000000   | 1     |
| 2022-01-02T00:00:00.000000   | 2     |
| 2022-01-03T00:00:00.000000   | 1     |
| 2022-01-04T00:00:00.000000   | 1     |
----------------------------------------

如何合并两个主题以及时间戳订购的结果？

Answer 1

您可以通过.option（“订阅”，“创建，删除”）读取两个主题

然后您都有两个主题的数据框能够按时间戳进行排序，并汇总/减少数据框架以获取输出。

另外，Kafka记录已经在其中有一个时间戳，火花返回为列。因此，您可以将生产者设计更改为具有一个主题，例如evest，从值中删除时间戳，将id移动到记录键创建事件将具有非null value，并且删除事件将具有null value。

无论哪种方式，您仍然需要一个还原函数。

---

或者，您无法使用SPARK，并且使用建议的主题设计，然后在Kafka Streams / ksqldb中创建一个表已经拥有您想要的数据。不一定使用时间戳信息，而是至少按ID或其他值进行汇总计数。