Spark流式处理时如何解析DataFrame中存储的复杂JSON字符串

实现思路 #

你需要依次完成「JSON字符串解析、动态键展开、字段提取」三个核心步骤，Spark 2.1.x内置的JSON解析函数和集合函数可以直接实现需求，无需额外依赖。

完整实现代码 #

#!/usr/bin/env python3
from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, explode, col
from pyspark.sql.types import *
import numpy as np

TCP_IP = 'localhost'
TCP_PORT = 6100

spark = SparkSession \
    .builder \
    .appName("Image Classifier") \
    .getOrCreate()

# 1. 定义JSON结构的Schema
# 最外层是字符串为key的Map，每个value对应单条样本的img和label
sample_schema = StructType([
    StructField("img", ArrayType(ArrayType(ArrayType(IntegerType())))),
    StructField("label", StringType())
])
json_schema = MapType(StringType(), sample_schema)

# 读取Socket流
dfFromStream = spark.readStream \
      .format("socket") \
      .option("host", TCP_IP) \
      .option("port", TCP_PORT) \
      .load()

# 2. 转换流程
# 2.1 解析JSON字符串为Map结构
parsed_df = dfFromStream.withColumn("parsed_json", from_json(col("value"), json_schema))
# 2.2 展开Map的键值对为多行，每个样本对应一行
exploded_df = parsed_df.select(explode(col("parsed_json")).alias("sample_id", "sample_data"))
# 2.3 提取img和label字段，得到目标格式
result_df = exploded_df.select(
    col("sample_data.img").alias("img"),
    col("sample_data.label").alias("label")
)

# 【可选】如果需要将Spark Array格式的img转为Numpy数组，可使用UDF
def array_to_numpy(arr):
    return np.array(arr)
numpy_udf = udf(array_to_numpy, BinaryType()) # 可根据后续处理需求调整返回类型
result_df_with_np = result_df.withColumn("img_np", numpy_udf(col("img")))

# 测试输出（可替换为自身业务的输出逻辑）
query = result_df.writeStream \
    .outputMode("append") \
    .format("console") \
    .start()
query.awaitTermination()

关键说明 #

• 三维数组的Schema需要严格匹配数据层级：ArrayType(ArrayType(ArrayType(IntegerType()))) 对应「三维嵌套数组」的结构，层级错误会导致解析结果为null
• 如果你的图片像素是0-255的无符号整数，可以把Schema里的IntegerType替换为ByteType减少存储占用
• Spark 2.1.x完全支持上述所有用到的内置函数，无需升级版本即可运行

内容的提问来源于stack exchange，提问作者Hari Raagav T R