spark - 当数据框中不存在列时设置 null

Question

我正在加载许多版本的 JSON 文件来激发 DataFrame。 一些文件包含 A、B 列 还有一些 A,B,C 或 A,C..

如果我运行这个命令

from pyspark.sql import SQLContext

sqlContext = SQLContext(sc)

df = sqlContext.sql("SELECT A,B,C FROM table")

加载几个后，我会收到错误“列不存在”我只加载了不包含 C 列的文件。 如何将此值设置为null 而不会出错？

Answer 1

DataFrameReader.json 方法提供了可以在此处使用的可选模式参数。如果您的架构很复杂，最简单的解决方案是重用从包含所有字段的文件中推断出的一个：

df_complete = spark.read.json("complete_file")
schema = df_complete.schema

df_with_missing = spark.read.json("df_with_missing", schema)
# or
# spark.read.schema(schema).("df_with_missing")

如果您知道架构，但由于某种原因无法在上面使用，则必须从头开始创建它。

schema = StructType([
    StructField("A", LongType(), True), ..., StructField("C", LongType(), True)])

一如既往，请务必在加载数据后执行一些质量检查。

示例（注意所有字段均为nullable）：

from pyspark.sql.types import *

schema = StructType([
    StructField("x1", FloatType()),
    StructField("x2", StructType([
        StructField("y1", DoubleType()),
        StructField("y2", StructType([
            StructField("z1", StringType()),
            StructField("z2", StringType())
        ]))
    ])),
    StructField("x3", StringType()),
    StructField("x4", IntegerType())
])

spark.read.json(sc.parallelize(["""{"x4": 1}"""]), schema).printSchema()
## root
##  |-- x1: float (nullable = true)
##  |-- x2: struct (nullable = true)
##  |    |-- y1: double (nullable = true)
##  |    |-- y2: struct (nullable = true)
##  |    |    |-- z1: string (nullable = true)
##  |    |    |-- z2: string (nullable = true)
##  |-- x3: string (nullable = true)
##  |-- x4: integer (nullable = true)

spark.read.json(sc.parallelize(["""{"x4": 1}"""]), schema).first()
## Row(x1=None, x2=None, x3=None, x4=1)

spark.read.json(sc.parallelize(["""{"x3": "foo", "x1": 1.0}"""]), schema).first()
## Row(x1=1.0, x2=None, x3='foo', x4=None)

spark.read.json(sc.parallelize(["""{"x2": {"y2": {"z2": "bar"}}}"""]), schema).first()
## Row(x1=None, x2=Row(y1=None, y2=Row(z1=None, z2='bar')), x3=None, x4=None)

重要：

此方法仅适用于 JSON 源，具体取决于实现细节。不要将其用于 Parquet 等来源。