如何在使用 Spark Streaming 流式传输 kafka 时对消息进行重复数据删除？答案

【问题标题】：How to deduplicate messages while streaming kafka using Spark Streaming?如何在使用 Spark Streaming 流式传输 kafka 时对消息进行重复数据删除？
【发布时间】：2018-07-22 05:13:13
【问题描述】：

我有一个案例，Kafka 生产者每天发送两次数据。这些生产者从数据库/文件中读取所有数据并发送到 Kafka。所以这些消息每天都发送，这是重复的。我需要使用 Spark Streaming 对消息进行重复数据删除并写入一些持久存储。在这种情况下，删除重复消息的最佳方法是什么？

发送的重复消息是一个带有时间戳字段的 json 字符串，仅更新。

注意： 我无法将 Kafka Producer 更改为仅发送新数据/消息，它已安装在客户端计算机中并由其他人编写.

【问题讨论】：

标签： apache-spark duplicates apache-kafka spark-streaming

【解决方案1】：

对于重复数据删除，您需要在某处存储有关已处理内容的信息（例如消息的唯一 ID）。

要存储您可以使用的消息：

火花检查点。优点：开箱即用。缺点：如果更新应用程序的源代码，则需要清理检查点。结果，您将丢失信息。如果重复数据删除的要求不严格，解决方案可以工作。
任何数据库。例如，如果您在 hadoop 环境上运行，则可以使用 Hbase。对于您“获取”的每条消息（检查它之前是否未发送），并在真正发送时在 DB 中标记。

【讨论】：

感谢您的回复，检查 id 将不正确，因为有时数据会使用相同的 id 更新。生成并检查 has 而不是 ids 是否是个好主意。
@ShankarKoirala 您可以存储 id + 时间戳。或整个消息的 md5 哈希。此逻辑不太依赖于您选择的解决方案。
我没用过。但是看不出它不应该工作的任何原因。确保在真正存储时标记消息发送。这可能会导致重复（例如重新启动时），但不会丢失数据。
如果数据更新并且我们检查的 id 相同，选项 2 会怎样。在这种情况下，数据将会丢失。

【解决方案2】：

您可以将主题配置更改为compact 模式。通过压缩，具有相同键的记录将在 Kafka 日志中被覆盖/更新。在那里，您只能从 Kafka 获得密钥的最新值。

您可以阅读有关压缩的更多信息here。

【讨论】：

据我了解，要启用代理范围的压缩，您必须在 server.properties 中设置 log.cleanup.policy=compact - 对吧？
是的，没错。但是有一个问题：一旦你设置了清理策略压缩，retention.ms 将不会被尊重。这是因为 kafka 清理线程要么根据用户设置的保留策略进行日志清理，要么进行压缩（dedup）。

【解决方案3】：

您可以尝试使用mapWithState。检查我的answer。

【讨论】：

【解决方案4】：

一个更简单的方法是在 kafka 端解决这个问题。看看 kafka 的 Log compaction 功能。如果记录具有相同的唯一键，它将为您删除重复记录。

https://kafka.apache.org/documentation/#compaction

【讨论】：

【解决方案5】：

您可以使用 Key-Value 数据存储，其中您的键将是不包括时间戳字段的字段组合和实际 json 的值。

当您轮询记录时，创建键和值对写入数据存储区，该数据存储区要么处理 UPSERT（插入 + 更新），要么检查数据存储区中是否存在密钥，然后删除消息

if(Datastore.get(key)){ 
     // then drop
 }else { 
    //write to the datastore
    Datastore.put(key)
}

我建议您检查 HBase（处理 UPSERTS）和 Redis（用于查找的内存中 KV 数据存储）

【讨论】：

【解决方案6】：

你有没有研究过这个： https://spark.apache.org/docs/latest/structured-streaming-programming-guide.html#streaming-deduplication

您可以尝试使用 dropDuplicates() 方法。如果您有多个列需要用于确定重复项，则可以使用 dropDuplicates(String[] colNames) 来传递它们。

【讨论】：