You've got two choices (and both require fairly intense coding which is OK, isn't it? :)).
Write your own custom Kafka Deserializer and use it where you use
StringDecoderin your example.
After you loaded a dataset (for a batch) convert it using
foreachoperator or use
maptransformation that applies the conversion as part of the pipeline.
You could also consider using spark-avro library.
- How to process messages in avro format from Kafka?
- How to deserialize Avro messages from Kafka in Flink (Scala)?
- How to write data in Avro format to Kafka from Flink?
- How to read records in JSON format from Kafka using Structured Streaming?
- Reading Avro messages from Kafka with Spark 2.0.2 (structured streaming)
- How do I read binary serialized Avro (Confluent Platform) from Kafka using Spark Streaming
- Scala: Error reading Kafka Avro messages from spark structured streaming
- How to process Scala case class objects from Kafka using streaming queries?
- Spark: Reading Avro messages from Kafka using Spark Scala
- Spark Streaming From Kafka and Write to HDFS in Avro Format
- How do you return a future containing a list of messages after all available messages have been consumed from a Kafka topic?
- How to only process messages from child actors?
- How to consume kafka events serialized to AVRO from Flink?
- How to parse Nested Json messages from Kafka topic to hive
- How to read records from Kafka topic from beginning in Spark Streaming?
- how to process data in chunks/batches with kafka streams?
- How SBT test task manages class path and how to correctly start a Java process from SBT test
- How to read json data using scala from kafka topic in apache spark
- How to gracefully terminate a process created from scala
- Kafka to S3 - How to loading slices from kafka to S3
- How to use kafka.group.id and checkpoints in spark 3.0 structured streaming to continue to read from Kafka where it left off after restart?
- How to stream data from Kafka topic to Delta table using Spark Structured Streaming
- Kafka MirrorMaker's consumer not fetching all messages from topics
- How to log all incoming messages from Akka (Java)
- How to write to Kafka from Spark with a changed schema without getting exceptions?
- How to get all sub-strings of a specific format from a string
- How do you send data from Kafka streams to clients via Websockets in PlayFramework?
- Read from Kafka topic process the data and write back to Kafka topic using scala and spark
- How to load and process multiple csv files from a DBFS directory with Spark
- How to get actor messages from stdin?
More Query from same tag
- Idiomatic table cell renderers in Scala
- Scala Actor receive definition with Spray
- use flatmap to filter and compute values in a collection
- Scala What is this "_1" in type?
- Scala: insert Long into BigInt using bit shifting
- How to use correctly mapPartitions function
- Generating two different random number simultaneously in Scala - Spark
- Elasticsearch Integration Tests Framework
- spark, scala & jdbc - how to limit number of records
- sbt stage does not include all configs from submodules
- MongoDB - Sorting by Reference or 'Foreign Key' (Liftweb, Scala)
- How to check empty element in json
- Play-slick - Is default.db required?
- How to read a chain of Scala function parameters
- JSON deserialization in scala - use default value if field doesn't exist
- How to get status of _delete_by_query api in elasticsearch using scala
- Split function difference between char and string arguments
- Scala: How do I get this toy covariant example to work?
- Why does Option(null).map result in an error?
- How do I split contents of a list that is part of a dataframe?
- Why does var cause a type variance error where val compiles in Scala?
- Scala collections contains not using ==?
- Akka Http Route Test: Request was neither completed nor rejected within 1 second
- How to use Scala to partition data into buckets for further processing
- Scala implicit conversions and parameters
- Convert csv file to map
- Getting actual type of Option[Any]
- Having hard time to understand why this function is called MapReduce
- Extractor conflicts in scala
- How is state managed in Scala Play! 2.0 Websockets?