Spark 2.4下PySpark读取ORC展平数据返回空值求助

Spark 2.4 与 Spark 3.x 的 from_json 函数在行为上存在明显差异，这是导致你的代码在2.4环境下解析后全为空值的核心原因。以下是针对性的排查方向和修复方案：

1. 字段名大小写敏感性问题 #

Spark 2.4 的 from_json 默认严格区分字段名大小写，而Spark 3.x对大小写的兼容性更强。如果content列中的JSON键与生成的Spark Schema字段名大小写不匹配（比如JSON里是user_id，Schema里是UserId），会直接导致解析失败返回空值。

修复方案：

• 确认Avro Schema中的字段名与JSON实际键名完全一致（包括大小写）
• 或通过配置关闭Spark的大小写敏感性：

  spark.conf.set("spark.sql.caseSensitive", "false")
  

2. Avro联合类型处理逻辑适配 #

你的代码处理Avro联合类型时直接取field_type[1]（默认第一个是null），但Spark 2.4对联合类型的解析逻辑更严格，需要明确处理null分支：

修复方案：
修改avro_schema_to_pyspark_schema函数，确保联合类型的解析逻辑适配Spark 2.4：

def avro_schema_to_pyspark_schema(avro_schema):
    '''
    将Schema Registry的Avro Schema转换为Spark Schema
    '''
    fields = avro_schema["fields"]
    struct_fields = []
    
    for field in fields:
        field_name = field["name"]
        field_type = field["type"]
        
        if isinstance(field_type, list):
            # 过滤Avro联合类型中的null，提取有效类型
            non_null_types = [t for t in field_type if t != "null"]
            if len(non_null_types) != 1:
                raise ValueError("不支持包含多个非null类型的联合类型")
            pyspark_type = avro_to_pyspark_type(non_null_types[0])
        else:
            pyspark_type = avro_to_pyspark_type(field_type)
        struct_fields.append(StructField(field_name, pyspark_type, nullable=True))
    return StructType(struct_fields)

3. 逻辑类型的兼容性修复 #

Spark 2.4对Avro逻辑类型的支持不如3.x完善，尤其是Decimal和Timestamp类型：

• Decimal类型：Spark 2.4需要明确指定精度和刻度，否则解析失败
• Timestamp-millis：Spark 2.4中用TimestampType替代LongType才能正确解析时间戳

修复方案：
更新avro_to_pyspark_type函数，完善逻辑类型处理：

def avro_to_pyspark_type(type_str):
    if isinstance(type_str, dict):
        if "logicalType" in type_str:
            logical_type = type_str["logicalType"]
            if logical_type == "date":
                return IntegerType()
            elif logical_type == "timestamp-millis":
                # Spark 2.4中用TimestampType适配毫秒级时间戳
                return TimestampType()
            elif logical_type == "decimal":
                # 从Avro Schema中提取精度和刻度参数
                precision = type_str.get("precision", 10)
                scale = type_str.get("scale", 0)
                return DecimalType(precision=precision, scale=scale)
            else:
                raise ValueError(f"不支持的Avro逻辑类型: {logical_type}")
        else:
            raise ValueError(f"不支持的复杂类型: {type_str}")
    elif type_str == "string":
        return StringType()
    elif type_str == "int":
        return IntegerType()
    elif type_str == "double":
        return DoubleType()
    elif type_str == "boolean":
        return BooleanType()
    elif type_str == "long":
        return LongType()
    elif type_str == "bytes":
        return BinaryType()
    elif type_str == "null":
        return NullType()
    else:
        raise ValueError(f"不支持的数据类型: {type_str}")

4. 宽松JSON格式适配 #

Spark 2.4的from_json对JSON格式要求更严格，不允许单引号、未加引号的键或 trailing commas。如果你的content列是类JSON而非标准JSON，需要显式配置兼容选项：

修复方案：
调用from_json时添加兼容选项：

df = df.withColumn(
    "Content", 
    from_json(
        col("Content"), 
        spark_schema, 
        options={
            "allowSingleQuotes": "true",
            "allowUnquotedFieldNames": "true",
            "ignoreTrailingCommas": "true"
        }
    )
)

5. 验证JSON格式有效性 #

先提取一条content列样本，确认JSON格式是否符合标准：

sample_json = df.select("Content").limit(1).collect()[0][0]
print(sample_json)

如果存在格式问题（比如单引号、缺失键引号），需要先清洗数据再解析。

内容的提问来源于stack exchange，提问作者Ali Lordifar