火花流为每个触发过程间隔的每条记录选择最新事件

如何解决火花流为每个触发过程间隔的每条记录选择最新事件

我们有一个火花流（火花版本 2.4.0）作业，它使用一个 Kafka 主题（4 个分区），其中包括带有 Id 的 json 形式的业务更改。 这些 Kafka 值还包括 RecordTime 字段和 json 对象中的其他字段。 此流式作业根据 Id 字段更新 Kudu 表。

一段时间后，我们注意到，有些更新实际上并未反映某些 id 字段值的最新状态。 我们假设每个分区有 4 个不同的 executor 处理，当其中一个比另一个更早完成时，它会更新目标 Kudu 表。 所以如果我们有如下值：

(Id=1,val=A,RecordTime: 10:00:05 ) partition1
(Id=2,RecordTime: 10:00:04 ) partition1
(Id=1,val=B,RecordTime: 10:00:07 ) partition2
(Id=1,val=C,RecordTime: 10:00:06 ) partition3
(Id=2,val=D,RecordTime: 10:00:06 ) partition4
(Id=1,val=E,RecordTime: 10:00:03 ) partition4

那么 Kudu 表应该是这样的：

Id	值	RecordTime
1	B	10:00:07
2	C	10:00:06

但是，有时我们会看到这样的 Kudu 表：

Id	值	RecordTime
1	A	10:00:05
2	C	10:00:06

触发间隔为 1 分钟。

那么，我们如何实现目标Kudu表的有序更新。

1. 我们是否应该使用单个分区进行排序，但如果我们这样做有好处/坏处吗？
2. 对于火花流，我们如何在每个触发间隔选择最新的记录和值
3. 根据 id 和 RecordTime 更新 kudu 表，但如何？
4. 我们还能想到其他方法吗？

希望我能充分解释我的问题。 简而言之，我们如何在 Spark 流中实现每个微批次间隔的事件排序？

特别感谢任何可以帮助我的人。

解决方法

当您从 Kafka 获取数据时，记住 Kafka 仅提供主题分区内的排序保证很有用。

因此，如果您让 Kafka 生产者将相同 ID 的所有消息生成到同一个分区中，您就可以解决您的问题。这可以通过 KafkaProducer 中的自定义分区器来实现，或者如果您只是使用 id 的值作为 Kafka 消息的“关键”部分。

如果您无法控制 Kafka 生产者，则需要使您的 Spark Streaming 作业有状态。在这里，具有挑战性的部分是定义一个时间范围，您的工作应等待具有相同 ID 的其他消息到达的时间。只是几秒钟吗？也许几个小时？我的经验是，这可能很难回答，有时答案是“几个小时”，这意味着您需要将状态保持几个小时，这可能会使您的工作超出内存。