Spark中DataFrame.show（）方法中的java.NullPointException - scala

Question

编辑：抱歉之前的问题质量，我希望这个问题会更清楚： 使用 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 状态如下： 数据框： 数据1（id）， 数据2（id）， 数据3（id）， 数据11（id）， 数据12（id）， md1(id), md2(id)

问题来了 - 如果目录中的 JSON 文件之一不包含 md2 节点，由于 NullPointException，我无法在“md2”数据帧上运行 show() 和 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")    

Answer 1

快速修复

尝试像这样插入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()

使用explode()

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

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()