title: covert Spark rdd[row] to dataframe
body: <p>I have trouble to transform json to dataframe.
  I am trying to use spark for a project to synchronize table to data lake（hudi） from a CDC(canal) listening mysql binlog. Here I received json about row change and add some fields for it. this json steam include multiple schema. each schemahave different colums and may add new column in the future.so I build GenericRowWithSchema for each json and pass individual schema for each row.</p>
  <p>Now, I need to tranform rdd[row] to dataframe to write to hudi How I can trans it?</p>
  <pre><code>object code{
  
  def main(args: Array[String]): Unit = {
  val sss = SparkSession.builder().appName(&quot;SparkHudi&quot;).getOrCreate()
  //val sc = SparkContext.getOrCreate
  val sc = sss.sparkContext
  val ssc = new StreamingContext(sc, Seconds(1))
  //ssc.sparkContext.setLogLevel(&quot;INFO&quot;);
  
  import org.apache.kafka.common.serialization.StringDeserializer
  val kafkaParams = Map[String, Object](
    ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -&gt; &quot;kafka.test.com:9092&quot;,
    ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer],
    ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG -&gt; classOf[StringDeserializer],
    ConsumerConfig.GROUP_ID_CONFIG -&gt; &quot;group-88&quot;,
    ConsumerConfig.AUTO_OFFSET_RESET_CONFIG -&gt; &quot;latest&quot;,
    ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG -&gt; Boolean.box(true)
  )
  
  val topics = Array(&quot;test&quot;)
  
  val kafkaDirectStream  = KafkaUtils.createDirectStream[String, String](
    ssc,
    LocationStrategies.PreferConsistent,
    ConsumerStrategies.Subscribe[String, String](topics, kafkaParams)
  )
  
  //Cache your RDD before you perform any heavyweight operations.
  
  kafkaDirectStream.start()
  
  
  val saveRdd = kafkaDirectStream.map(x =&gt; {
    //receive json from kafka
    val jsonObject = JSON.parse(x.value()).asInstanceOf[JSONObject]
    jsonObject
  }).filter(json =&gt;{
  
  
    /*some json field operation*/
  
    val keySets = dataJson.keySet()
  
    val dataArray:ArrayBuffer[AnyRef] = ArrayBuffer[AnyRef]()
    val fieldArray:ArrayBuffer[StructField] = ArrayBuffer[StructField]()
  
  
    keySets.forEach(dataKey=&gt;{
      
  fieldArray.append(
   StructField(dataKey,SqlTypeConverter.toSparkType(sqlTypeJson.getIntValue(dataKey))))
  
      dataArray.append(dataJson.get(dataKey));
    })
  
    val schema = StructType(fieldArray)
    val row = new GenericRowWithSchema(dataArray.toArray, schema).asInstanceOf[Row]
  
    row
  })
  
  
  
  
  saveRdd.foreachRDD ( rdd =&gt; {
    // Get the offset ranges in the RDD
    //println(rdd.map(x =&gt; x.toJSONString()).toDebugString());
  
    sss.implicits
  
  
  
  
    rdd.collect().foreach(x=&gt;{
      println(x.json)
      println(x.schema.sql)
    })
  
  })
  </code></pre>