Json Parsing 在 Spark UDF 中抛出意外的输出答案

【问题标题】：Json Parsing throws unexpected output inside Spark UDFJson Parsing 在 Spark UDF 中抛出意外的输出
【发布时间】：2018-03-04 23:35:18
【问题描述】：

我有一个数据框。

该数据框中所有列的数据类型都是 string 。有些列是 jsonString

 +--------+---------+--------------------------+
 |event_id|event_key|              rights      |
 +--------+---------+--------------------------+
 |     410|(default)|{"conditions":[{"devic...|
 +--------+---------+--------------------------+

我想单独解析该 jsonString 并从中获取一个值并将其添加为新列。我正在使用 Jackson 解析器来做到这一点。

这里是“权利”的价值

 {
"conditions": [
    {
        "devices": [
            {
                "connection": [
                    "BROADBAND",
                    "MOBILE"
                ],
                "platform": "IOS",
                "type": "MOBILE",
                "provider": "TELETV"
            },
            {
                "connection": [
                    "BROADBAND",
                    "MOBILE"
                ],
                "platform": "ANDROID",
                "type": "MOBILE",
                "provider": "TELETV"
            },
            {
                "connection": [
                    "BROADBAND",
                    "MOBILE"
                ],
                "platform": "IOS",
                "type": "TABLET",
                "provider": "TELETV"
            },
            {
                "connection": [
                    "BROADBAND",
                    "MOBILE"
                ],
                "platform": "ANDROID",
                "type": "TABLET",
                "provider": "TELETV"
            }
        ],
        "endDateTime": "2017-01-09T22:59:59.000Z",
        "inclusiveGeoTerritories": [
            "DE",
            "IT",
            "ZZ"
        ],
        "mediaType": "Linear",
        "offers": [
            {
                "endDateTime": "2017-01-09T22:59:59.000Z",
                "isRestartable": true,
                "isRecordable": true,
                "isCUTVable": false,
                "recordingMode": "UNIQUE",
                "retentionCUTV": "P7DT2H",
                "retentionNPVR": "P2Y6M5DT12H35M30S",
                "offerId": "MOTOGP-RACE",
                "offerType": "IPPV",
                "startDateTime": "2017-01-09T17:00:00.000Z"
            }
        ],
        "platformName": "USA",
        "startDateTime": "2017-01-09T17:00:00.000Z",
        "territory": "USA"
    }
 ]
}

现在我想在现有数据框中创建一个新列。要添加的新列的名称是“提供者”

 conditions -> devices -> provider

我想对 dataframe 中的每一行执行此操作。因此，我创建了一个 UDF，并且我将保存 jsonString 的列传递给该 udf，并且在该 udf 内部我想解析 json 字符串并需要以字符串形式返回值

我的火花代码：

 import org.apache.spark.sql.functions.udf
 import org.apache.spark.sql.functions._
 import org.json4s._
 import org.json4s.jackson.JsonMethods
 import org.json4s.jackson.JsonMethods._


  //
     some codes to derive base dataframe
  //

  val fetchProvider_udf = udf(fetchProvider _)
  val result = df.withColumn("provider",fetchProvider_udf(col("rights")))
   result.select("event_id","event_key","rights","provider").show(10)


  def fetchProvider(jsonStr:String): String = {

    val json = JsonMethods.parse(jsonStr)

   val providerData = json \\ "conditions" \\"devices" \\ "provider"

   compact(render(providerData))
  }

此外，如果导航键不可用，我该如何处理？它会抛出异常吗？可以说“条件”存在并且“设备”存在但“提供者”键在 json 字符串中不存在。那我该如何处理呢？

谁能帮帮我

预期输出：

 +--------+---------+-----------------------+-------------+
 |event_id|event_key|              rights     |provider     |
 +--------+---------+-----------------------+-------------+
 |     410|(unknown)|{"conditions":[{"devic...|    TELETV    |
 +--------+---------+-----------------------+-------------+

但我得到以下输出

 +--------+---------+-----------------------+-------------------------------     ------------------------------------------------------+
 |event_id|event_key|              rights        |                                                     provider     |
      +--------+---------+-----------------------+--------------------------      -----------------------------------------------------------+
 |     410|(unknown)|{"conditions":[{"devic...|    {"provider":"TELETV","provider":"TELETV","provider":"TELETV","provider":"TELETV"      }   |
   +--------+---------+-----------------------+-----------------------------       --------------------------------------------------------+

【问题讨论】：

有什么理由不用spark的get_json_object？
要求是使用任何任何scala解析器。

标签： json parsing apache-spark

【解决方案1】：

如果您想提取第一个提供者的值，您应该在 UDF 中使用以下代码：

(json \\ "conditions" \\"devices")[0] \\ "provider"

当前代码只是获取所有提供者（作为 Map），然后将其转换为字符串作为 UDF 结果。

您还应该确保您的 UDF 不会引发任何异常（因为它会导致整个作业失败）。最简单的方法是返回 null 然后：

如果你想调查 - 过滤df.provider.isNull()
如果您只想保留有效条目 - 按df.provider.isNullNull() 过滤

【讨论】：