PySpark：使用嵌套数组值作为列名动态创建DataFrame

完全可以实现，批处理和流处理场景都适用，具体思路和代码如下：

实现思路 #

• 先定义输入JSON的结构Schema，解析原始数据得到包含headers和data两个字段的DataFrame
• 使用explode函数将data字段里的二维数组展开，每个子数组对应一行
• 按位置提取展开后的数组元素，映射到headers对应的列名，同时可以按需转换字段类型

可运行代码示例 #

以下为批处理场景的样例代码，流处理场景仅需替换数据源读取逻辑即可：

from pyspark.sql import SparkSession
from pyspark.sql.functions import explode, col, from_json
from pyspark.sql.types import StructType, ArrayType, StringType

if __name__ == "__main__":
    spark = SparkSession.builder.appName("JsonFlatten").getOrCreate()
    
    # 1. 定义输入JSON的Schema
    input_schema = StructType() \
        .add("headers", ArrayType(StringType())) \
        .add("data", ArrayType(ArrayType(StringType())))
    
    # 2. 读取原始JSON数据，替换为你的实际数据源路径
    df = spark.read.schema(input_schema).json("your_input_path.json")
    
    # 3. 获取固定列名（前置条件保证所有headers一致，取第一条即可）
    col_names = df.first()["headers"]
    
    # 4. 展开data数组，映射为结构化列
    result_df = df.select(explode("data").alias("row_data")) \
        .select(*[col("row_data")[i].alias(col_names[i]) for i in range(len(col_names))])
    
    # 可选：按需转换字段类型，比如转成DoubleType
    # result_df = result_df.select(
    #     col("col_a").cast("double"),
    #     col("col_b").cast("double"),
    #     col("col_c").cast("double"),
    #     col("col_d").cast("double")
    # )
    
    # 验证输出
    result_df.show()
    spark.stop()

流式场景适配说明 #

如果是流式输入，仅需将第2步的读取逻辑替换为流式读取即可，后续转换逻辑完全一致，以Kafka数据源为例：

df = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "your_kafka_server") \
    .option("subscribe", "your_topic") \
    .load() \
    .select(col("value").cast("string").alias("json_str")) \
    .select(from_json("json_str", input_schema).alias("parsed_data")) \
    .select("parsed_data.*")

你已经提前清洗数据保证headers统一、data子数组长度匹配，上述代码可以稳定运行，不需要额外做异常兼容。

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