RDDからKafkaOffsetにアクセス中の例外

Question

私は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 => { }) 

dataStrea誰かが私が私がここで間違ってやっているかを理解することができた場合

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

：mはダイレクト・ストリームである（DSTREAM [文字列]）のようなKafkaUtilsのAPIのものを使用して作成しました。同様

おかげで公式ドキュメントで述べたように 変換データストリーム上で実行する方法のチェーンの最初の方法があります。

Answer 1

あなたの問題がある：代わりにKafkaUtils.createKafkaStreamによって作成されたDirectKafkaInputDStreamのMapPartitionedDStreamを作成

.map(._2) 

。

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

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