如何在PySpark中解析列行分离的JSON响应并完成列行映射

步骤1：解析JSON字符串，提取核心数据 #

先通过Python的json模块解析原始响应字符串，拆分出列定义和行数据：

import json
from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StructField, IntegerType, StringType, TimestampType

# 原始响应字符串
response_str = '{"columns": [{ "id": { "type": "Numeric", "nullable": false } },{ "name": { "type": "Text", "nullable": false } },{ "description": { "type": "Text", "nullable": true } },{ "last_updated": { "type": "DateTime", "nullable": false } }],"rows": [[1, "foo", "Lorem ipsum", "2016-10-26T00:09:14Z"],[4, "bar", null, "2013-07-01T13:04:24Z"]]}'

# 解析JSON结构
response_data = json.loads(response_str)
columns_def = response_data['columns']
rows_data = response_data['rows']

步骤2：将列定义转换为PySpark Schema #

把JSON中的类型标识映射为PySpark支持的数据类型，构建结构化Schema：

# 定义类型映射规则
type_mapping = {
    "Numeric": IntegerType(),
    "Text": StringType(),
    "DateTime": TimestampType()
}

# 生成Schema字段
schema_fields = []
for col_item in columns_def:
    # 提取列名（每个字典仅含一个键）
    col_name = next(iter(col_item.keys()))
    col_details = col_item[col_name]
    # 匹配PySpark数据类型
    data_type = type_mapping[col_details['type']]
    nullable = col_details['nullable']
    schema_fields.append(StructField(col_name, data_type, nullable=nullable))

spark_schema = StructType(schema_fields)

步骤3：创建映射完成的DataFrame #

用解析后的行数据和Schema生成最终的PySpark DataFrame：

# 初始化SparkSession
spark = SparkSession.builder.appName("ColRowMapping").getOrCreate()

# 构建DataFrame
df = spark.createDataFrame(rows_data, schema=spark_schema)

# 验证结果
df.show()
df.printSchema()

运行后会得到列名、数据类型完全匹配的DataFrame，其中last_updated会被识别为时间戳类型，description允许为null值。

额外说明 #

• 如果Numeric类型对应超大数值，可将映射改为LongType()或DecimalType()适配
• ISO格式的时间字符串可直接被TimestampType解析，无需额外转换
• 若JSON来自外部文件，可先用spark.read.json()加载后再提取columns和rows字段，重复上述流程

内容的提问来源于stack exchange，提问作者Sreenivasa Reddy Gopireddy