私はこのドキュメントに従いましたが、うまくいきました。ここで、spark からコネクタ データを使用しようとしました。私が使用できる参照はありますか?私はコンフルエントを使用しているため、元のカフカのリファレンス ドキュメントとは大きく異なります。
これまでに使用したコードです。問題は、レコード データを java.String に変換できないことです。(そして、それが消費する正しい方法であるかどうかはわかりません)
val brokers = "http://127.0.0.1:9092"
val topics = List("postgres-accounts2")
val sparkConf = new SparkConf().setAppName("KafkaWordCount")
//sparkConf.setMaster("spark://sda1:7077,sda2:7077")
sparkConf.setMaster("local[2]")
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
sparkConf.registerKryoClasses(Array(classOf[org.apache.avro.generic.GenericData$Record]))
val ssc = new StreamingContext(sparkConf, Seconds(2))
ssc.checkpoint("checkpoint")
// Create direct kafka stream with brokers and topics
//val topicsSet = topics.split(",")
val kafkaParams = Map[String, Object](
"schema.registry.url" -> "http://127.0.0.1:8081",
"bootstrap.servers" -> "http://127.0.0.1:9092",
"key.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer",
"value.deserializer" -> "io.confluent.kafka.serializers.KafkaAvroDeserializer",
"group.id" -> "use_a_separate_group_id_for_each_stream",
"auto.offset.reset" -> "earliest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val messages = KafkaUtils.createDirectStream[String, String](
ssc,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)
val data = messages.map(record => {
println( record)
println( "value : " + record.value().toString() ) // error java.lang.ClassCastException: org.apache.avro.generic.GenericData$Record cannot be cast to java.lang.String
//println( Json.parse( record.value() + ""))
(record.key, record.value)
})