我正在加载多个版本的 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
而不是出错?
最佳答案
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)])
一如既往,一定要在加载数据后执行一些质量检查。
示例(注意所有字段都是可为空
):
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 等来源。
关于python - spark - 当数据框中不存在列时设置为空,我们在Stack Overflow上找到一个类似的问题: https://stackoverflow.com/questions/32166812/