在第一批处理后关闭Spark流式处理上下文（尝试检索kafka偏移量）

Question

我正在尝试为我的Spark Batch工作检索Kafka补偿。在检索偏移量之后，我想关闭流上下文。

我尝试将streamlistener添加到流上下文，并在作业完成后实现onBatchCompleted方法关闭流，但收到异常“无法停止侦听器总线线程内的StreamingContext”。

有解决办法吗？我正在尝试检索偏移量以调用KafkaUtils。createRDD（sparkContext、kafkaProperties、OffsetRange[]，LocationStrateg）

private OffsetRange[] getOffsets(SparkConf sparkConf) throws InterruptedException {
    final AtomicReference<OffsetRange[]> atomicReference = new AtomicReference<>();

    JavaStreamingContext sc = new JavaStreamingContext(sparkConf, Duration.apply(50));
    JavaInputDStream<ConsumerRecord<String, String>> stream =
            KafkaUtils.createDirectStream(sc, LocationStrategies.PreferConsistent(), ConsumerStrategies.<String, String>Subscribe(Arrays.asList("test"), getKafkaParam()));
    stream.foreachRDD((VoidFunction<JavaRDD<ConsumerRecord<String, String>>>) rdd -> {
                atomicReference.set(((HasOffsetRanges) rdd.rdd()).offsetRanges());
                // sc.stop(false); //this would throw exception saying consumer is already closed
            }
    );
    sc.addStreamingListener(new TopicListener(sc)); //Throws exception saying "Cannot stop StreamingContext within listener bus thread."
    sc.start();
    sc.awaitTermination();
    return atomicReference.get();
}



public class TopicListener implements StreamingListener {
private JavaStreamingContext sc;

public TopicListener(JavaStreamingContext sc){
    this.sc = sc;
}
@Override
public void onBatchCompleted(StreamingListenerBatchCompleted streamingListenerBatchCompleted) {
    sc.stop(false);
}

非常感谢Stackers:）我曾尝试寻找可能的解决方案，但迄今为止没有成功

编辑：我使用KafkaConsumer来获取分区信息。获得分区信息后，我创建一个TopicPartition pojo列表，并调用position和endOffsets方法分别获取我的groupId的当前位置和end位置。

final List<PartitionInfo> partitionInfos = kafkaConsumer.partitionsFor("theTopicName");
final List<TopicPartition> topicPartitions = new ArrayList<>();
partitionInfos.forEach(partitionInfo -> topicPartitions.add(new TopicPartition("theTopicName", partitionInfo.partition())));
final List<OffsetRange> offsetRanges = new ArrayList<>();
kafkaConsumer.assign(topicPartitions);
topicPartitions.foreach(topicPartition -> {
    long fromOffset = kafkaConsumer.position(topicPartition);
    kafkaConsumer.seekToEnd(Collections.singleton(topicPartition));
    long untilOffset = kafkaConsumer.position(topicPartition);
    offsetRanges.add(new OffsetRange(topicPartition.topic(), topicPartition.partition(), fromOffset, untilOffset));
});
return offsetRanges.toArray(new OffsetRange[offsetRanges.size()]);

Answer 1

如果你想控制流，你可以考虑使用轮询而不是流api。这样，一旦你的目标达到，你就可以清楚地停止投票。

另外看看这个...

https://github.com/dibbhatt/kafka-spark-consumer