title: Deserialize self-referencing protobuf in Spark Structured Streaming
body: <p>I have a self-referencing protobuf schema:</p>
  
  <pre><code>message A { 
   uint64 timestamp = 1; 
   repeated A fields = 2; 
  }
  </code></pre>
  
  <p>I am generating the corresponding Scala classes using <code>scalaPB</code> and then trying to decode the messages which are consumed from Kafka stream, following these steps:</p>
  
  <pre><code>def main(args : Array[String]) {
  
    val spark = SparkSession.builder.
      master("local")
      .appName("spark session example")
      .getOrCreate()
  
    import spark.implicits._
  
    val ds1 = spark.readStream.format("kafka").
        option("kafka.bootstrap.servers","localhost:9092").
        option("subscribe","student").load()
  
    val ds2 = ds1.map(row=&gt; row.getAs[Array[Byte]]("value")).map(Student.parseFrom(_))
  
    val query = ds2.writeStream
        .outputMode("append")
        .format("console")
        .start()
  
    query.awaitTermination()
  
  }
  </code></pre>
  
  <p><a href="https://stackoverflow.com/questions/40649109/error-when-decoding-the-proto-buf-messages-in-spark-streaming-using-scalapb">This is a related question here on StackOverflow</a>. </p>
  
  <p>However, Spark Structured Streaming throws a cyclic reference error at this line.</p>
  
  <pre><code>val ds2 = ds1.map(row=&gt; row.getAs[Array[Byte]]("value")).map(Student.parseFrom(_))
  </code></pre>
  
  <p>I understand it is because of the recursive reference which can be handled in the Spark only at the driver (basically <code>RDD</code> or <code>Dataset</code> level). Has anyone figured a workaround for this, to enable recursive calling through UDF for instance?</p>