title: How to deserialize records from Kafka using Structured Streaming in Java?
body: <p>I use Spark <strong>2.1</strong>.</p>
  
  <p>I am trying to read records from Kafka using Spark Structured Streaming, deserialize them and apply aggregations afterwards.</p>
  
  <p>I have the following code:</p>
  
  <pre class="lang-java prettyprint-override"><code>SparkSession spark = SparkSession
          .builder()
          .appName("Statistics")
          .getOrCreate();
  
  Dataset&lt;Row&gt; df = spark
          .readStream()
          .format("kafka")
          .option("kafka.bootstrap.servers", kafkaUri)
          .option("subscribe", "Statistics")
          .option("startingOffsets", "earliest")
          .load();
  
  df.selectExpr("CAST(value AS STRING)")
  </code></pre>
  
  <p>What I want is to deserialize the <code>value</code> field into my object instead of casting as <code>String</code>.</p>
  
  <p>I have a custom deserializer for this.</p>
  
  <pre><code>public StatisticsRecord deserialize(String s, byte[] bytes)
  </code></pre>
  
  <p>How can I do this in Java?</p>
  
  <hr>
  
  <p>The only relevant link I have found is this <a href="https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html" rel="nofollow noreferrer">https://databricks.com/blog/2017/04/26/processing-data-in-apache-kafka-with-structured-streaming-in-apache-spark-2-2.html</a>, but this is for Scala.</p>