由于数据类型不匹配而获取:参数2需要整数类型解析Json数据时出错Spark SQL [duplicate]

Question

我试图解析JSON数据，为此我编写了自定义模式。通过添加模式或不添加模式来解析数据时，我得到以下错误:

Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve '`queryResults`.`searchResponse`.`response`.`docs`.`transactions`['code']' due to data type mismatch: argument 2 requires integral type, however, ''code'' is of string type.;;

这是我的示例数据：

{
    "queryResults": {
        "searchResponse": {
            "response": {
                "docs": [{
                    "transactions": [{
                        "recordDate": "2010-02-02 00:00:00",
                        "code": "PGM/",
                        "description": "Recordation of Patent Grant Mailed"
                    }, {
                        "recordDate": "2010-01-13 00:00:00",
                        "code": "WPIR",
                        "description": "Issue Notification Mailed"
                    }, {
                        "recordDate": "2009-12-17 00:00:00",
                        "code": "R1021",
                        "description": "Receipt into Pubs"
                    }]
                }]
            }
        }
    }
}

这是我的模式：

val schema=StructType(List(
      StructField("queryResults",StructType(
        List(StructField("searchResponse",StructType(
          List(StructField("response",StructType(
            List(StructField("docs",ArrayType(StructType(
              List(
                StructField("appCustNumber", StringType, nullable = true),
                StructField("transactions",ArrayType(StructType(
                  List
                  (
                    StructField("code", StringType, nullable = true),
                    StructField("description", StringType, nullable = true),
                    StructField("recordDate", StringType, nullable = true)
                  )
                )))
              )
            ))))
          )))
        )))
      ))
    ))

以下是我尝试获取数据的方式:

val dff = sqlCotext.read.schema(schema).json("file locatiuon")
dff.select("queryResults.searchResponse.response.docs.transactions.code").show()

提前致谢。

Answer 1

试试这个...

val dfContent = dff.select(explode(dff("queryResults.searchResponse.response.docs.transactions"))).toDF("transaction")

val code = dfContent.select("transaction.code")

code.show(false)