如何在 JSON 中使用 read.schema 仅指定特定字段：SPARK Scala

Question

我正在尝试以编程方式在看起来像 json 的文本文件上强制实施架构 (json)。我尝试使用 jsonFile 但问题是从 json 文件列表创建数据框，spark 必须对数据进行 1 次传递才能为数据框创建模式。因此它需要解析所有花费较长时间的数据（自从我的数据压缩后 4 小时，大小为 TB）。因此，我想尝试将其读取为文本文件并强制执行架构以单独获取感兴趣的字段，以便稍后查询生成的数据框。但我不确定如何将其映射到输入。有人可以给我一些关于如何将模式映射到 json 的参考，比如输入。

输入：

这是完整的架构：

records: org.apache.spark.sql.DataFrame = [country: string, countryFeatures: string, customerId: string, homeCountry: string, homeCountryFeatures: string, places: array<struct<freeTrial:boolean,placeId:string,placeRating:bigint>>, siteName: string, siteId: string, siteTypeId: string, Timestamp: bigint, Timezone: string, countryId: string, pageId: string, homeId: string, pageType: string, model: string, requestId: string, sessionId: string, inputs: array<struct<inputName:string,inputType:string,inputId:string,offerType:string,originalRating:bigint,processed:boolean,rating:bigint,score:double,methodId:string>>]

但我只对几个领域感兴趣，例如：

res45: Array[String] = Array({"requestId":"bnjinmm","siteName":"bueller","pageType":"ad","model":"prepare","inputs":[{"methodId":"436136582","inputType":"US","processed":true,"rating":0,"originalRating":1},{"methodId":"23232322","inputType":"UK","processed":falase,"rating":0,"originalRating":1}]


 val  records = sc.textFile("s3://testData/sample.json.gz")

  val schema = StructType(Array(StructField("requestId",StringType,true),
                          StructField("siteName",StringType,true),
                          StructField("model",StringType,true),
                          StructField("pageType",StringType,true),
                          StructField("inputs", ArrayType(
                                StructType(
                                            StructField("inputType",StringType,true), 
                                            StructField("originalRating",LongType,true), 
                                            StructField("processed",BooleanType,true), 
                                            StructField("rating",LongType,true), 
                                            StructField("methodId",StringType,true)
                                            ),true),true)))

    val rowRDD = ?? 

    val inputRDD = sqlContext.applySchema(rowRDD, schema)
    inputRDD.registerTempTable("input")

     sql("select * from input").foreach(println)

有什么方法可以映射这个吗？或者我是否需要使用子解析器或其他东西。由于限制，我只想使用 textFile。

尝试过：

val  records =sqlContext.read.schema(schema).json("s3://testData/test2.gz")

但不断收到错误：

<console>:37: error: overloaded method value apply with alternatives:
     (fields: Array[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType <and>
      (fields: java.util.List[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType <and>
      (fields: Seq[org.apache.spark.sql.types.StructField])org.apache.spark.sql.types.StructType
     cannot be applied to (org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField, org.apache.spark.sql.types.StructField)
           StructField("inputs",ArrayType(StructType(StructField("inputType",StringType,true), StructField("originalRating",LongType,true), StructField("processed",BooleanType,true), StructField("rating",LongType,true), StructField("score",DoubleType,true), StructField("methodId",StringType,true)),true),true)))
                                              ^

Answer 1

它可以使用预定义模式加载以下代码，spark 不需要通过 ZIP 文件中的文件。问题中的代码有歧义。

import org.apache.spark.sql.types._

val input = StructType(
                Array(
                    StructField("inputType",StringType,true), 
                    StructField("originalRating",LongType,true), 
                    StructField("processed",BooleanType,true), 
                    StructField("rating",LongType,true), 
                    StructField("score",DoubleType,true), 
                    StructField("methodId",StringType,true)
                )
            )

 val schema = StructType(Array(
    StructField("requestId",StringType,true),
    StructField("siteName",StringType,true),
    StructField("model",StringType,true),
    StructField("inputs",
        ArrayType(input,true),
                true)
    )
)

val  records =sqlContext.read.schema(schema).json("s3://testData/test2.gz")

并非所有字段都需要提供。虽然如果可能的话最好提供所有。

如果某些行无效，Spark 会尽力解析所有行。它将添加 _corrupt_record 作为包含整行的列。而如果它是平原 json 文件文件。

如何在 JSON 中使用 read.schema 仅指定特定字段：SPARK Scala

How to specify only particular fields using read.schema in JSON : SPARK Scala

json

scala

apache-spark

rdd