decoding Avro messages in a spark scala streaming code.

Anonymous — Sat, 16 Nov 2024 08:32:07 GMT

Hi,

I'm trying to stream unstructured encoded avro message data from a kafka topic via scala. normal message (not avro encoded) is working fine. Problem is decoding avro encoded message. Any help will be fine. Code part is below:

import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark._
import org.apache.spark.sql.SparkSession
import org.apache.spark.streaming._
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.kafka010._

object Consumer4 {
def main(args: Array[String]): Unit = {

val spark = SparkSession
.builder()
.appName("Kafka test")
.getOrCreate();

val sparkContext = SparkContext.getOrCreate()
val streamingContext = new StreamingContext(sparkContext, Seconds(2))
val preferredHosts = LocationStrategies.PreferConsistent
val topics = List("xyz") //

val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "xyz.xyz.com.tr:1111",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[StringDeserializer],
"group.id" -> "use_a_separate_group_id_for_each_stream",
"auto.offset.reset" -> "latest",
"enable.auto.commit" -> (false: java.lang.Boolean)
)
val stream = KafkaUtils.createDirectStream[String, String](
streamingContext,
PreferConsistent,
Subscribe[String, String](topics, kafkaParams)
)
// stream.map(record => (record.key, record.value))
stream.foreachRDD { rdd =>
// val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
rdd.foreachPartition { iter =>

while (iter.hasNext) {
val item = iter.next()
println(item)
}

// val o: OffsetRange = offsetRanges(TaskContext.get.partitionId)
// println(s"${o.topic} ${o.partition} ${o.fromOffset} ${o.untilOffset}")
}
}
streamingContext.start()
streamingContext.awaitTermination()
}
}

Re: decoding Avro messages in a spark scala streaming code.

Anonymous — Tue, 20 Mar 2018 07:47:43 GMT

There should be at least suggestions or advise etc..

Some help will be great and i will appreciate it 🙂

Re: decoding Avro messages in a spark scala streaming code.

Anonymous — Mon, 26 Mar 2018 02:57:03 GMT

is the above code is being implemented through Talend ? problem statement is not clear, please explain ?

topic decoding Avro messages in a spark scala streaming code. in Talend Studio

decoding Avro messages in a spark scala streaming code.

Re: decoding Avro messages in a spark scala streaming code.

Re: decoding Avro messages in a spark scala streaming code.