如何在 Spark Kafka 直接流中手动提交偏移量？

Question

我环顾四周，但没有找到满意的答案。也许我错过了一些东西。请帮忙。

我们有一个使用 Kafka 主题的 Spark 流应用程序，它需要确保在推进 Kafka 偏移量之前进行端到端处理，例如更新数据库。这很像在流系统中构建事务支持，并保证每条消息都被处理（转换），更重要的是，输出。

我已阅读有关 Kafka DirectStreams 的信息。它说为了在 DirectStreaming 模式下进行稳健的故障恢复，应该启用 Spark 检查点，它将偏移量与检查点一起存储。但是偏移管理是在内部完成的（设置 Kafka 配置参数，如 [ "auto.offset.reset", "auto.commit.enable", "auto.offset.interval.ms"]）。它没有说明我们如何（或是否）可以自定义提交偏移量（例如，一旦我们加载了数据库）。换句话说，我们可以设置"auto.commit.enable"为 false 并自己管理偏移量（与数据库连接不同）吗？

非常感谢任何指导/帮助。

Answer 1

下面的文章可能是理解该方法的良好开端。

Spark-Kafka-实现零数据丢失

此外，

文章建议直接使用zookeeper客户端，也可以用KafkaSimpleConsumer之类的东西代替。使用 Zookeper/KafkaSimpleConsumer 的优点是依赖于 Zookeper 保存的偏移量的监控工具。此外，信息还可以保存在 HDFS 或任何其他可靠的服务上。