1

编辑：对以前的问题质量感到抱歉，我希望这个问题会更清楚：使用 Spark 应用程序，我正在加载以下 JSON 文件的整个目录：

    {
        "type": "some_type",
        "payload": {
            "data1": {
                "id": "1"           
            },
            "data2": {
                "id": "1",

            },
            "data3": {
                "id": "1"
            },
            "dataset1": [{
                "data11": {
                    "id": "1",
                },
                "data12": {
                    "id": "1",
                }
            }],
            "masterdata": {
                "md1": [{
                    "id": "1"
                },
                {
                    "id": "2"
                },
                {
                    "id": "3"
                }],
                "md2": [{
                    "id": "1",
                },
                {
                    "id": "2",
                },
                {
                    "id": "3",
                }]
            }
        }
    }

进入DataFrame并保存为临时表，以便以后使用。在此 Json 中，“有效负载”节点中的字段始终存在，但“主数据”中的子节点是可选的。下一步是为 Json 的每个子节点创建多个 DataFrame，如下所示：DataFrame data1 包含来自所有文件的节点“data1”的数据，看起来像一个带有“id”列的常规表。在第一个处理部分之后，我的 Spark 状态如下： DataFrames: data1(id), data2(id), data3(id), data11(id), data12(id), md1(id), md2(id)

问题来了 - 如果目录中的一个 JSON 文件不包含 md2 节点，由于 NullPointException ，我既不能运行也show()不能在“md2”DataFrame 上运行。collect()我会理解是否所有文件都缺少“md2”节点，因此它无法创建 md2 DataFrame，但在这种情况下，我希望 md2 DataFrame 根本没有来自没有节点 md2 的 json 文件的数据，但包含所有其他文件。

技术细节：要从嵌套节点读取数据，我使用 rdd.map 和 rdd.flatmap，然后我将其转换为DataFrame自定义列名

如果我在目录中的所有文件包含所有节点时运行应用程序，一切正常，但是如果单个文件丢失 md2 节点应用程序在 .show() 或 .collect() 上失败

顺便说一句，如果节点存在但它为空，则一切正常。

有什么方法可以让 Spark 支持可选的 Json 节点或处理 rdd.map&flatmap 中缺失的节点？

我希望它比上一个问题更清楚

在@Beryllium 请求中，这里是我用来获取 md2 DataFrame 的 rdd 操作

    val jsonData = hiveContext.sql("SELECT `payload`.masterdata.md2 FROM jsonData")
    val data = jsonData.rdd.flatMap(row => row.getSeq[Row](0)).map(row => (
    row.getString(row.fieldIndex("id"))
    )).distinct
    val dataDF = data.toDF("id")
4

1 回答 1

3

快速解决

尝试插入filter()这样的：

sqlContext.sql("SELECT payload.masterdata.md2 FROM jsonData")
  .rdd
  .filter(_.getSeq[Row](0) != null)
  .flatMap(row => row.getSeq[Row](0))
  .map(row => (row.getString(row.fieldIndex("id"))))
  .distinct
  .toDF("id")
  .show()

使用爆炸（）

这会尽快删除空值：所以它应该更快（至少它更短）：

sqlContext
  .sql("select t.a.id from (SELECT explode(payload.masterdata.md2) as a FROM jsonData) t")
  • explode()爆炸了null。
  • 然后子查询只提取ID

更简单：先提取 ID，然后explode()：

sqlContext.sql("SELECT explode(payload.masterdata.md2.id) FROM jsonData").show()
于 2015-11-27T11:23:58.990 回答