как использовать spark.read. json для нескольких json столбцов из фрейма данных - PullRequest
Новая
       208

как использовать spark.read. json для нескольких json столбцов из фрейма данных

0 голосов
/ 07 августа 2020

Итак, вместо одного столбца, как нам изменить код для чтения из нескольких json столбцов?

в настоящее время используется только столбец «col Json» из фрейма данных. но список столбцов нужно читать аналогичным образом. Список столбцов хранится в переменной List [String]. 

      val data = Seq(
        (77, "email1", """{"key1":38,"key3":39}"""),
        (78, "email2", """{"key1":38,"key4":39}"""),
        (178, "email21", """{"key1":"when string","key4":36, "key6":"test", "key10":false }"""),
        (179, "email8", """{"sub1":"qwerty","sub2":["42"]}"""),
          (180, "email8", """{"sub1":"qwerty","sub2":["42", "56", "test"]}""")
      ).toDF("id", "name", "colJson")
    
      data.show(false)
    //  +---+-------+---------------------------------------------------------------+
    //  |id |name   |colJson                                                        |
    //  +---+-------+---------------------------------------------------------------+
    //  |77 |email1 |{"key1":38,"key3":39}                                          |
    //  |78 |email2 |{"key1":38,"key4":39}                                          |
    //  |178|email21|{"key1":"when string","key4":36, "key6":"test", "key10":false }|
    //  |178|email8 |{"sub1":"qwerty","sub2":"42"}                                  |
    //  +---+-------+---------------------------------------------------------------+
    
    
      val schema = spark.read.json(data.select("colJson").as[String]).schema
      val res = data.select($"id", $"name", from_json($"colJson", schema).as("s")).select("id", "name", "s.*")
      res.show(false)
    //  +---+-------+-----------+-----+----+----+----+------+----+
    //  |id |name   |key1       |key10|key3|key4|key6|sub1  |sub2|
    //  +---+-------+-----------+-----+----+----+----+------+----+
    //  |77 |email1 |38         |null |39  |null|null|null  |null|
    //  |78 |email2 |38         |null |null|39  |null|null  |null|
    //  |178|email21|when string|false|null|36  |test|null  |null|
    //  |178|email8 |null       |null |null|null|null|qwerty|42  |
    //  +---+-------+-----------+-----+----+----+----+------+----+
    
      val  df1 = res.filter('sub1.equalTo("qwerty"))
      df1.show(false)
    //  +---+------+----+-----+----+----+----+------+----+
    //  |id |name  |key1|key10|key3|key4|key6|sub1  |sub2|
    //  +---+------+----+-----+----+----+----+------+----+
    //  |178|email8|null|null |null|null|null|qwerty|42  |
    //  +---+------+----+-----+----+----+----+------+----+

1 Ответ

1 голос
/ 07 августа 2020

Проверьте код ниже.

Добавлен еще один столбец с данными json. 

scala> val df = Seq(
    (77, "email1", """{"key1":38,"key3":39}""","""{"name":"aaa","age":10}"""),
    (78, "email2", """{"key1":38,"key4":39}""","""{"name":"bbb","age":20}"""),
    (178, "email21", """{"key1":"when string","key4":36, "key6":"test", "key10":false }""","""{"name":"ccc","age":30}"""),
    (179, "email8", """{"sub1":"qwerty","sub2":["42"]}""","""{"name":"ddd","age":40}"""),
    (180, "email8", """{"sub1":"qwerty","sub2":["42", "56", "test"]}""","""{"name":"eee","age":50}""")
).toDF("id", "name", "colJson","personInfo")

scala> df.printSchema
root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = true)
 |-- colJson: string (nullable = true)
 |-- personInfo: string (nullable = true)

scala> df.show(false)
+---+-------+---------------------------------------------------------------+-----------------------+
|id |name   |colJson                                                        |personInfo             |
+---+-------+---------------------------------------------------------------+-----------------------+
|77 |email1 |{"key1":38,"key3":39}                                          |{"name":"aaa","age":10}|
|78 |email2 |{"key1":38,"key4":39}                                          |{"name":"bbb","age":20}|
|178|email21|{"key1":"when string","key4":36, "key6":"test", "key10":false }|{"name":"ccc","age":30}|
|179|email8 |{"sub1":"qwerty","sub2":["42"]}                                |{"name":"ddd","age":40}|
|180|email8 |{"sub1":"qwerty","sub2":["42", "56", "test"]}                  |{"name":"eee","age":50}|
+---+-------+---------------------------------------------------------------+-----------------------+

создано из Json неявной функции , Вы можете передать в это несколько столбцов, и он будет анализировать и извлекать столбцы из json. 

scala> :paste
// Entering paste mode (ctrl-D to finish)

import org.apache.spark.sql.{Column, DataFrame, Row}
    import org.apache.spark.sql.functions.from_json
    implicit class DFHelper(inDF: DataFrame) {
      import inDF.sparkSession.implicits._
      def fromJson(columns:Column*):DataFrame = {
        val schemas = columns.map(column => (column, inDF.sparkSession.read.json(inDF.select(column).as[String]).schema))
        val mdf = schemas.foldLeft(inDF)((df,schema) => {
                df.withColumn(schema._1.toString(),from_json(schema._1,schema._2))
        })        
        mdf.selectExpr(mdf.schema.map(c => if(c.dataType.typeName =="struct") s"${c.name}.*" else c.name):_*)
      }
    }

// Exiting paste mode, now interpreting.

import org.apache.spark.sql.{Column, DataFrame, Row}
import org.apache.spark.sql.functions.from_json
defined class DFHelper

scala> df.fromJson($"colJson",$"personInfo").show(false)

+---+-------+-----------+-----+----+----+----+------+--------------+---+----+
|id |name   |key1       |key10|key3|key4|key6|sub1  |sub2          |age|name|
+---+-------+-----------+-----+----+----+----+------+--------------+---+----+
|77 |email1 |38         |null |39  |null|null|null  |null          |10 |aaa |
|78 |email2 |38         |null |null|39  |null|null  |null          |20 |bbb |
|178|email21|when string|false|null|36  |test|null  |null          |30 |ccc |
|179|email8 |null       |null |null|null|null|qwerty|[42]          |40 |ddd |
|180|email8 |null       |null |null|null|null|qwerty|[42, 56, test]|50 |eee |
+---+-------+-----------+-----+----+----+----+------+--------------+---+----+

scala> df.fromJson($"colJson",$"personInfo").printSchema()
root
 |-- id: integer (nullable = false)
 |-- name: string (nullable = true)
 |-- key1: string (nullable = true)
 |-- key10: boolean (nullable = true)
 |-- key3: long (nullable = true)
 |-- key4: long (nullable = true)
 |-- key6: string (nullable = true)
 |-- sub1: string (nullable = true)
 |-- sub2: array (nullable = true)
 |    |-- element: string (containsNull = true)
 |-- age: long (nullable = true)
 |-- name: string (nullable = true)
...