Spark Scala DataFrame单行转换为JSON用于PostrgeSQL插入

Question

使用名为< code>lastTail的数据帧，我可以这样迭代:

import scalikejdbc._
// ... 
// Do Kafka Streaming to create DataFrame lastTail
// ...

lastTail.printSchema

lastTail.foreachPartition(iter => {

// 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
  }
})

这将输出“类似”的内容（带编校）：root|--fileGid:string（nullable=true）|--eventStruct:struct（nullaable=false）||--event Index:integer（nullable=true）||--eventGid:string（nullable=true）| |--eventType:string||--eventType:string（nullable=true）

和（只有一个迭代项-经过了编辑，但希望有足够好的语法）

**** 类组织 apache.spark.sql.catalyst.expressions.GenericRow与Schema 12345 模式： 结构类型（结构字段（文件gid，字符串类型，true）， 结构字段（事件结构，结构类型（事件索引，整数类型，true）， 结构字段（事件Gid，字符串类型，true）， 结构字段（事件类型，字符串类型，true））， 结构字段（修订结构，结构类型（结构字段（事件索引，整数类型，true），结构字段（事件类型，字符串类型，true），结构字段（事件类型，字符串类型，true）， 结构字段（事件类型，字符串类型，true），字符串： [12345，[1，4，编辑]，[1，4，修订版]

注意：我在 https://github.com/koeninger/kafka-exactly-once/blob/master/src/main/scala/example/TransactionalPerPartition.scala 上做像价值指标= iter.sum这样的部分，而是使用数据帧。我还遵循 http://spark.apache.org/docs/latest/streaming-programming-guide.html#performance-tuning 中看到的“使用前切RDD的设计模式”。

如何将这个org . Apache . spark . SQL . catalyst . expressions . genericrowwithschema(见https://github . com/Apache/spark/blob/master/SQL/catalyst/src/main/Scala/org/Apache/spark/SQL/catalyst/expressions/rows . Scala)迭代项转换成一个容易编写的东西(JSON或...？-我是开放的)进入PostgreSQL。(如果不是JSON，请建议如何将这个值读回到DataFrame中，以便在其他地方使用。)

Answer 1

好吧，我想出了一种不同的方法来解决这个问题。

val ltk = lastTail.select($"fileGid").rdd.map(fileGid => fileGid.toString)
val ltv = lastTail.toJSON
val kvPair = ltk.zip(ltv)

然后，我会简单地迭代RDD，而不是数据帧。

kvPair.foreachPartition(iter => {
  while(iter.hasNext) {
    val item = iter.next()
    println(item.getClass)
    println(item)
  }
})

撇开数据不谈，我得到了< code >类scala。Tuple2，这使得在JDBC / PostgreSQL中存储KV对变得更加容易。

我敢肯定，还有其他不可行的方法。