PySpark DataFrame提取固定字段 剩余字段转为Map列实现方法

核心实现逻辑 #

• 先明确所有需要提取为独立列的固定公共字段列表
• 将原始JSON数据统一解析为支持键值访问的结构
• 单独提取固定字段作为独立列
• 过滤掉固定字段后，将剩余动态字段统一组装为Map类型列

场景1：从Kafka/Socket等读取原始JSON字符串流 #

这种场景下读入的原始数据为JSON格式字符串，可直接将整条JSON解析为Map结构处理，无需提前感知所有动态字段，适配性最强。
完整实现代码：

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, from_json, map_filter
from pyspark.sql.types import MapType, StringType

# 初始化Spark会话
spark = SparkSession.builder.appName("StreamFieldProcess").getOrCreate()

# 定义需要提取的固定公共字段
fixed_fields = ["event1", "event2", "event3"]

# 读取Kafka流式数据，其他文本类流数据源逻辑一致
raw_stream = spark.readStream \
    .format("kafka") \
    .option("kafka.bootstrap.servers", "localhost:9092") \
    .option("subscribe", "test_topic") \
    .load()

# 将Kafka中二进制类型的value转为字符串JSON
json_base_df = raw_stream.selectExpr("CAST(value AS STRING) as raw_json")

# 将整条JSON解析为Map类型列，键值均为字符串类型（如果有混合类型可调整value类型）
parsed_df = json_base_df.withColumn(
    "json_kv",
    from_json(col("raw_json"), MapType(StringType(), StringType()))
)

# 提取固定字段为独立列
for field in fixed_fields:
    parsed_df = parsed_df.withColumn(field, col("json_kv").getItem(field))

# 过滤掉固定字段，剩余键值对存入extra_fields列
result_df = parsed_df.withColumn(
    "extra_fields",
    map_filter(col("json_kv"), lambda k, v: ~k.isin(*fixed_fields))
).drop("raw_json", "json_kv") # 删除中间临时列

# 启动流任务，此处以控制台输出为例，生产环境替换为对应sink即可
query = result_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()
query.awaitTermination()

针对给出的样例输入，处理后输出结构如下：

场景2：直接用Spark内置JSON源读取流式文件 #

如果是用spark.readStream.json()直接读取目录下的JSON文件，Spark会自动将所有顶级JSON字段拆分为独立列，此时可通过列名差集组装Map列，实现代码片段如下：

from pyspark.sql.functions import col, lit, create_map

# 直接读取JSON文件流，Spark自动解析所有字段为列
json_stream = spark.readStream.json("/path/to/your/json/dir")

# 计算非固定字段列表
all_columns = json_stream.columns
extra_columns = [c for c in all_columns if c not in fixed_fields]

# 组装Map列：交替传入列名（作为key）、列值（作为value）
map_build_args = []
for col_name in extra_columns:
    map_build_args.append(lit(col_name))
    map_build_args.append(col(col_name))
extra_map_col = create_map(*map_build_args).alias("extra_fields")

# 组装最终结果
result_df = json_stream.select(*[col(f) for f in fixed_fields], extra_map_col)

注意事项 #

• 如果JSON字段值存在数值、布尔、嵌套结构等混合类型，可将解析Map的value类型替换为对应类型，Spark 3.4+版本可使用VariantType支持任意类型值
• 固定字段如果存在嵌套结构，可调整getItem的访问路径提取对应值，逻辑和顶级字段一致
• 批处理场景下上述代码完全通用，无需修改逻辑

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