Spark Scala DataFrame Single Row conversion to JSON for PostrgeSQL Insertion
<p>With a DataFrame called <code>lastTail</code>, I can iterate like this:</p>

<pre><code>import scalikejdbc._
// ... 
// Do Kafka Streaming to create DataFrame lastTail
// ...

lastTail.printSchema

lastTail.foreachPartition(iter =&gt; {

// open database connection from connection pool
// with scalikeJDBC (to PostgreSQL) 

  while(iter.hasNext) {
    val item = iter.next()
    println("****")
    println(item.getClass)
    println(item.getAs("fileGid"))
    println("Schema: "+item.schema)
    println("String: "+item.toString())
    println("Seqnce: "+item.toSeq)

    // convert this item into an XXX format (like JSON)
    // write row to DB in the selected format
  }
})
</code></pre>

<p>This outputs "something like" (with redaction):
<code>
   root
    |-- fileGid: string (nullable = true)
    |-- eventStruct: struct (nullable = false)
    |    |-- eventIndex: integer (nullable = true)
    |    |-- eventGid: string (nullable = true)
    |    |-- eventType: string (nullable = true)
    |-- revisionStruct: struct (nullable = false)
    |    |-- eventIndex: integer (nullable = true)
    |    |-- eventGid: string (nullable = true)
    |    |-- eventType: string (nullable = true)
</code></p>

<p>and (with just one iteration item - redacted,  but hopefully with good enough syntax as well)</p>

<p><code>****
    class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
    12345
    Schema: StructType(StructField(fileGid,StringType,true), StructField(eventStruct,StructType(StructField(eventIndex,IntegerType,true), StructField(eventGid,StringType,true), StructField(eventType,StringType,true)), StructField(revisionStruct,StructType(StructField(eventIndex,IntegerType,true), StructField(eventGid,StringType,true), StructField(eventType,StringType,true), StructField(editIndex,IntegerType,true)),false))
    String: [12345,[1,4,edit],[1,4,revision]]
    Seqnce: WrappedArray(12345, [1,4,edit], [1,4,revision])
</code></p>

<p>Note: I doing the part like <code>val metric = iter.sum</code> on <a href="https://github.com/koeninger/kafka-exactly-once/blob/master/src/main/scala/example/TransactionalPerPartition.scala" rel="nofollow">https://github.com/koeninger/kafka-exactly-once/blob/master/src/main/scala/example/TransactionalPerPartition.scala</a>, but with DataFrames instead. I am also following "Design Patterns for using foreachRDD" seen at  <a href="http://spark.apache.org/docs/latest/streaming-programming-guide.html#performance-tuning" rel="nofollow">http://spark.apache.org/docs/latest/streaming-programming-guide.html#performance-tuning</a>.</p>

<p>How can I convert this 
org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema 
(see <a href="https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/rows.scala" rel="nofollow">https://github.com/apache/spark/blob/master/sql/catalyst/src/main/scala/org/apache/spark/sql/catalyst/expressions/rows.scala</a>)
iteration item into a something that is easily written (JSON or ...? - I'm open) into PostgreSQL. (If not JSON, please suggest how to read this value back into a DataFrame for use at another point.)</p>