暂停并恢复SparkStreaming中的KafkaConsumer

Question

:)

我已经在一个（奇怪的）情况中结束了自己，简单地说，我不想使用来自Kafka的任何新记录，因此暂停主题中所有分区的sparkStreaming消费（InputStream[ConsumerRecord]），执行一些操作，最后，恢复消费记录。

首先这可能吗？

我一直在尝试这样的事情：

var consumer: KafkaConsumer[String, String] = _    
consumer = new KafkaConsumer[String, String](properties)    
consumer.subscribe(java.util.Arrays.asList(topicName))

consumer.pause(consumer.assignment())
...
consumer.resume(consumer.assignment())

但是我得到了这个：

println(s"Assigned partitions: $consumer.assignment()") --> []
println(s"Paused partitions: ${consumer.paused()}") --> []
println(s"Partitions for: ${consumer.partitionsFor(topicNAme)}") --> [Partition(topic=topicAAA, partition=0, leader=1, replicas=[1,2,3], partition=1, ... ]

任何帮助我理解我遗漏了什么，以及为什么当消费者明确分配了分区时我会得到空结果的帮助都将受到欢迎！

版本：Kafka：0.10火花：2.3。0斯卡拉：2.11。8.

Answer 1

是的，可以在代码中添加检查点并传递持久存储（本地磁盘、S3、HDFS）路径

每当您开始/恢复工作时，它将从检查点获取带有消费者偏移的Kafka消费者组信息，并从停止的地方开始处理。

val context = StreamingContext.getOrCreate(checkpointDirectory, functionToCreateContext _)

Spark Check-=定点不仅是用于保存偏移量的机制，而且还用于保存阶段和作业的DAG的序列化状态。因此，每当您使用新代码重新启动作业时，它都会

1. 读取并处理序列化数据
2. 如果您的Spark App中有任何代码更改，请清理缓存的DAG阶段
3. 用最新代码恢复处理新数据。

现在，从磁盘读取数据只是Spark加载Kafka偏移量、DAG和旧的未完成处理数据所需的一次性操作。

一旦完成，它将始终按照默认或指定的检查点间隔将数据保存到磁盘。

Spark streaming提供了指定Kafka组id的选项，但Spark structured stream没有。