kafka 主题中的唯一消息检查

我们使用 Logstash，希望从 Oracle 数据库读取一张表并将这些消息（如下所示）发送到 Kafka：

Topic1: message1: {"name":"name-1", "id":"fbd89256-12gh-10og-etdgn1234njF", "site":"site-1", "time":"2019-07-30"}
        message2: {"name":"name-2", "id":"fbd89256-12gh-10og-etdgn1234njG", "site":"site-1", "time":"2019-07-30"}
        message3: {"name":"name-3", "id":"fbd89256-12gh-10og-etdgn1234njS", "site":"site-1", "time":"2019-07-30"}
        message4: {"name":"name-4", "id":"fbd89256-12gh-10og-etdgn1234njF", "site":"site-1", "time":"2019-07-30"}

请注意message1 and message4是重复项same ID number.

现在，我们希望确保所有消息都是唯一的，那么我们如何过滤topic1并且所有消息都是唯一的，然后发送到topic2?

我们想要的最终结果：

Topic2: message1: {"name":"name-1", "id":"fbd89256-12gh-10og-etdgn1234njF", "site":"site-1", "time":"2019-07-30"}
        message2: {"name":"name-2", "id":"fbd89256-12gh-10og-etdgn1234njG", "site":"site-1", "time":"2019-07-30"}
        message3: {"name":"name-3", "id":"fbd89256-12gh-10og-etdgn1234njS", "site":"site-1", "time":"

这被称为一次性处理.

您可能对第一部分感兴趣卡夫卡常见问题解答 https://cwiki.apache.org/confluence/display/KAFKA/FAQ#FAQ-HowdoIgetexactly-oncemessagingfromKafka?描述了如何避免数据生产重复的一些方法（即从生产者方面）：

Exactly Once 语义有两部分：避免数据重复 生产并避免数据消费过程中的重复。

有两种方法可以在数据期间获取恰好一次语义 生产：

1. 每次获得网络时，每个分区都使用单写入器 错误检查该分区中的最后一条消息，看看您的最后一条消息是否 写入成功
2. 在中包含主键（UUID 或其他） 消息并对消费者进行重复数据删除。

如果您执行其中一项操作，Kafka 托管的日志将是 无重复。然而，无重复的阅读取决于一些 也来自消费者的合作。如果消费者定期 检查其位置，如果失败并重新启动，它将 从检查点位置重新启动。因此，如果数据输出和 检查点不是以原子方式写入的，因此可以获得 此处也重复。这个问题是你的存储所特有的 系统。例如，如果您正在使用数据库，您可以提交 这些一起在一个交易中。 LinkedIn 推出的 HDFS 加载器 Camus write 对 Hadoop 负载做了类似的事情。另一种选择 不需要事务的方法是将偏移量存储在 使用主题/分区/偏移量加载数据并进行重复数据删除 组合。

我认为有两个改进可以让这变得更容易：

1. 生产者幂等性可以自动完成并且成本更低 通过选择在服务器上集成对此的支持。
2. 现有的 高级消费者不会暴露很多更细粒度的信息 控制偏移（例如重置您的位置）。我们将工作 很快

另一个选择（这不完全是您正在寻找的）是日志压缩。假设您的重复消息具有相同的密钥，那么当日志压缩策略有效时，日志压缩最终将删除重复项。