小编tar*_*i43的帖子

我有一个来自Kafka的Spark消费者.我试图管理完全一次语义的偏移.

但是,在访问偏移量时会抛出以下异常:

"java.lang.ClassCastException:org.apache.spark.rdd.MapPartitionsRDD无法强制转换为org.apache.spark.streaming.kafka.HasOffsetRanges"

执行此操作的代码部分如下所示:

var offsetRanges = Array[OffsetRange]()
dataStream
  .transform { 
    rdd =>
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
   }
   .foreachRDD(rdd => { })

这里dataStream是使用KafkaUtils API创建的直接流(DStream [String]),如:

KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](ssc, kafkaParams, Set(source_schema+"_"+t)).map(_._2)

如果有人能帮助我理解我在这里做错了什么.transform是在官方文档中提到的数据流上执行的方法链中的第一个方法

谢谢.