apache-spark - 以许可模式读取 json 文件 - PySpark 2.3

Question

我有一个数据作业来读取一堆 json 文件，其中某些文件中的少数 json 行可能已损坏（无效的 json）。下面是代码：

df = spark.read \
        .option("mode", "PERMISSIVE")\
        .option("columnNameOfCorruptRecord", "_corrupt_record")\
        .json("hdfs://someLocation/")

对我来说发生的事情是，如果我尝试使用上述代码读取一个完全完美的文件（没有损坏的记录），则根本不会添加此列。

我的要求是添加这个“_corrupt_record”列，不管json文件是否有损坏的记录。如果文件没有任何损坏的记录，则该字段的所有值都应为空。

标签： apache-sparkpysparkapache-spark-sqlpyspark-dataframes

Answer 1

您可以检查该_corrupt_record列是否存在于 中df，如果不存在则手动添加。

import pyspark.sql.functions as F

if '_corrupt_record' not in df.columns:
    df = df.withColumn('_corrupt_record', F.lit(None))