スパーク[JSONObject]をデータセットにRDDする方法

Question

com.google.gson.JsonObject型のElementのRDDからデータを読み取り中です。これをDataSetに変換しようとしていますが、これを行う方法がわかりません。

import com.google.gson.{JsonParser} 
import org.apache.hadoop.io.LongWritable 
import org.apache.spark.sql.{SparkSession} 

object tmp { 
    class people(name: String, age: Long, phone: String) 
    def main(args: Array[String]): Unit = { 
    val spark = SparkSession.builder().master("local[*]").getOrCreate() 
    val sc = spark.sparkContext 

    val parser = new JsonParser(); 
    val jsonObject1 = parser.parse("""{"name":"abc","age":23,"phone":"0208"}""").getAsJsonObject() 
    val jsonObject2 = parser.parse("""{"name":"xyz","age":33}""").getAsJsonObject() 

    val PairRDD = sc.parallelize(List(
     (new LongWritable(1l), jsonObject1), 
     (new LongWritable(2l), jsonObject2) 
    )) 

    val rdd1 =PairRDD.map(element => element._2) 

    import spark.implicits._ 

    //How to create Dataset as schema People from rdd1? 
    } 
} 

がさえ要素RDD1印刷しようとすると、

object not serializable (class: org.apache.hadoop.io.LongWritable, value: 1) 
- field (class: scala.Tuple2, name: _1, type: class java.lang.Object) 
- object (class scala.Tuple2, (1,{"name":"abc","age":23,"phone":"0208"})) 

をスロー基本的に私は私が形質転換のためのSQLを適用することができるようにデータセットに変換したいBigQueryテーブルからこのRDD [LongWritable、JsonParser]を取得します。

私は意図的に2番目のレコードnullに電話を残しましたが、BigQueryはnull値を持つ要素に対して何も返しません。

Answer 1

ご清聴ありがとうございます。 KryoでSerializableとしてクラスを登録する必要があります。次のショーの仕事。それはより多くのアイデアを提供した場合、私はそう古いコンテキストを破棄し、登録Kryoクラス

import com.google.gson.{JsonParser} 
import org.apache.hadoop.io.LongWritable 
import org.apache.spark.SparkContext 

sc.stop() 

val conf = sc.getConf 
conf.registerKryoClasses(Array(classOf[LongWritable], classOf[JsonParser])) 
conf.get("spark.kryo.classesToRegister") 

val sc = new SparkContext(conf) 

val parser = new JsonParser(); 
val jsonObject1 = parser.parse("""{"name":"abc","age":23,"phone":"0208"}""").getAsJsonObject() 
val jsonObject2 = parser.parse("""{"name":"xyz","age":33}""").getAsJsonObject() 

val pairRDD = sc.parallelize(List(
    (new LongWritable(1l), jsonObject1), 
    (new LongWritable(2l), jsonObject2) 
)) 


val rdd = pairRDD.map(element => element._2) 

rdd.collect() 
// res9: Array[com.google.gson.JsonObject] = Array({"name":"abc","age":23,"phone":"0208"}, {"name":"xyz","age":33}) 

val jsonstrs = rdd.map(e=>e.toString).collect() 
val df = spark.read.json(sc.parallelize(jsonstrs))  
df.printSchema 
// root 
// |-- age: long (nullable = true) 
// |-- name: string (nullable = true) 
// |-- phone: string (nullable = true)