PySpark结构化流中Nullable=false设置失效，如何正确读取Schema？

我正在使用PySpark结构化流从Kafka读取数据，并尝试通过Struct schema转换JSON payload。

提供的JSON Schema如下：

{
    "fields": [
        {
            "metadata": {},
            "name": "test",
            "nullable": true,
            "type": {
                "containsNull": true,
                "elementType": {
                    "fields": [
                        {
                            "metadata": {},
                            "name": "message",
                            "nullable": false,
                            "type": "string"
                        },
                        {
                            "metadata": {},
                            "name": "recipient_id",
                            "nullable": true,
                            "type": "long"
                        }
                    ],
                    "type": "struct"
                },
                "type": "array"
            }
        },
        {
            "metadata": {},
            "name": "user_id",
            "nullable": true,
            "type": "long"
        }
    ],
    "type": "struct"
}

通过StructType.fromJson(jsonSchema)将该JSON Schema转换为StructType，结果如下：

StructType.fromJson(jsonSchema)

StructType([StructField('test', ArrayType(StructType([StructField('message', StringType(), False), StructField('recipient_id', LongType(), True)]), True), True), StructField('user_id', LongType(), True)])

使用该Schema转换数据后，生成的DataFrame Schema中，原本设置为nullable=false的字段仍显示为true，且向该字段传入null值未触发任何错误。转换代码如下：

spark_df = spark_df.selectExpr('timestamp', "CAST(value AS STRING)")
spark_df = spark_df.withColumn("value",from_json(col("value"),schemaNew, {"mode": "FAILFAST"}))
spark_df.printSchema()

输出的Schema：

root
 |-- timestamp: timestamp (nullable = true)
 |-- value: struct (nullable = true)
 |    |-- test: array (nullable = true)
 |    |    |-- element: struct (containsNull = true)
 |    |    |    |-- message: string (nullable = true)
 |    |    |    |-- recipient_id: long (nullable = true)
 |    |-- user_id: long (nullable = true)

请问如何从文件读取Schema并正确应用nullable属性，将JSON数据转换为符合预期的DataFrame？

核心问题分析 #

问题出在数组的containsNull配置：你的JSON Schema中test字段的ArrayType设置了containsNull: true，这允许数组内的结构体元素为null。此时Spark会优先遵循数组的null约束，忽略结构体内部字段的nullable=false设置——因为整个结构体元素都可以是null，内部字段的非空校验自然不会触发。

具体解决步骤 #

1. 修正JSON Schema的containsNull配置 #

将数组的containsNull改为false，确保数组中的每个结构体元素都非空，这样内部字段的nullable约束才能生效。修改后的完整JSON Schema如下：

{
    "fields": [
        {
            "metadata": {},
            "name": "test",
            "nullable": true,
            "type": {
                "containsNull": false,  // 关键修改：从true改为false
                "elementType": {
                    "fields": [
                        {
                            "metadata": {},
                            "name": "message",
                            "nullable": false,
                            "type": "string"
                        },
                        {
                            "metadata": {},
                            "name": "recipient_id",
                            "nullable": true,
                            "type": "long"
                        }
                    ],
                    "type": "struct"
                },
                "type": "array"
            }
        },
        {
            "metadata": {},
            "name": "user_id",
            "nullable": true,
            "type": "long"
        }
    ],
    "type": "struct"
}

2. 从文件读取Schema并应用 #

如果Schema存储在本地文件中，使用以下代码读取并转换为StructType，再应用到数据转换：

import json
from pyspark.sql.types import StructType
from pyspark.sql.functions import from_json, col

# 读取JSON Schema文件
with open("your_schema_file.json", "r") as schema_file:
    schema_json = json.load(schema_file)

# 转换为PySpark StructType
schema_new = StructType.fromJson(schema_json)

# 处理Kafka数据：转换value字段为JSON结构体
spark_df = spark_df.selectExpr('timestamp', "CAST(value AS STRING)")
spark_df = spark_df.withColumn("value", from_json(col("value"), schema_new, {"mode": "FAILFAST"}))

# 打印验证Schema
spark_df.printSchema()

此时输出的Schema会正确保留message字段的nullable=false属性：

root
 |-- timestamp: timestamp (nullable = true)
 |-- value: struct (nullable = true)
 |    |-- test: array (nullable = true)
 |    |    |-- element: struct (containsNull = false)
 |    |    |    |-- message: string (nullable = false)
 |    |    |    |-- recipient_id: long (nullable = true)
 |    |-- user_id: long (nullable = true)

3. 验证约束生效 #

使用以下测试数据验证：

• 合法数据：{"test": [{"message": "hello", "recipient_id": 123}], "user_id": 456}
• 非法数据（message为null）：{"test": [{"message": null, "recipient_id": 123}], "user_id": 456}
• 非法数据（数组元素为null）：{"test": [null], "user_id": 456}

在FAILFAST模式下，非法数据会立即抛出AnalysisException，符合预期的非空校验逻辑。

内容的提问来源于stack exchange，提问作者Mike Reddington