scala – 从RDD访问KafkaOffset时出现异常

发布时间：2020-12-16 10:05:38 所属栏目：安全来源：网络整理

导读：我有一个来自Kafka的Spark消费者. 我试图管理完全一次语义的偏移. 但是,在访问偏移量时会抛出以下异常： “java.lang.ClassCastException: org.apache.spark.rdd.MapPartitionsRDD cannot be cast to org.apache.spark.streaming.kafka.HasOffsetRanges” 执

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

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

“java.lang.ClassCastException: org.apache.spark.rdd.MapPartitionsRDD
cannot be cast to 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是在官方文档中提到的数据流上执行的方法链中的第一个方法

谢谢.

解决方法

你的问题是：

.map(._2)

这将创建一个MapPartitionedDStream,而不是由KafkaUtils.createKafkaStream创建的DirectKafkaInputDStream.

你需要在变换后进行映射：

val kafkaStream = KafkaUtils.createDirectStream[String,Set(source_schema+""+t))

kafkaStream
  .transform { 
    rdd => 
      offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
      rdd
  }
  .map(_._2)
  .foreachRDD(rdd => // stuff)

（编辑：李大同）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!