当 Spark 消费者从中读取时，Kafka 主题没有被刷新答案

【问题标题】：Kafka Topic not getting flushed when Spark consumer reads from it当 Spark 消费者从中读取时，Kafka 主题没有被刷新
【发布时间】：2019-01-12 13:46:59
【问题描述】：

我在 Spark 中使用以下消费者代码来读取 Kafka 主题：

val df = spark
  .read
  .format("kafka")
  .option("kafka.bootstrap.servers", kafkaBrokers)
  .option("subscribe", topicName)
  .load()
df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)")
  .as[(String, String)]

代码按预期从主题中读取，但主题的内容并没有因为这次读取而被刷新。重复执行会导致同一组消息一遍又一遍地返回。

我应该怎么做才能使消息在阅读时从主题中删除？

【问题讨论】：

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

【解决方案1】：

正如crikcet_007 提到的，Kafka 在消费后不会删除日志。您可以使用基于大小的策略或基于时间的设置来管理 Kafka 中的日志保留。

log.retention.bytes - 删除前日志的最大大小

log.retention.hours - 在删除之前保留日志文件的小时数

log.retention.minutes - 保存日志文件的分钟数

log.retention.ms - 保存日志文件的毫秒数

你可以阅读更多关于这些参数here

在处理日志保留的附加机制之上是日志压缩。通过设置以下参数，您可以管理日志压缩

log.cleanup.policy

log.cleaner.min.compaction.lag.ms

你可以阅读更多关于here

【讨论】：

【解决方案2】：

Kafka 在消费时不会删除主题消息

您的 Spark 代码是 Kafka 消费者组的一部分，它需要确认一条消息已被读取，并提交这些偏移量，我相信 Spark默认情况下会定期自动执行此操作，但您可以通过将enable.auto.commit 选项设置为false 来禁用此功能，强烈建议您这样做，因为您需要控制 Spark 是否已成功处理记录集合。

Checkpointing 或将偏移量提交到持久存储是 some ways 以在重新启动/任务失败的情况下保留您的偏移量，而不是重新读取相同的数据

【讨论】：