如何在PySpark DataFrame中高效拆分复杂地址字符串并展开数据

针对TB级数据场景，绝对不能用逗号拆分这种不靠谱的方式，必须用PySpark原生的JSON解析+数组展开方案，完全适配分布式大数据处理，不会因为地址内部的逗号出错。

核心思路 #

利用PySpark内置的from_json解析JSON格式的数组字符串，将其转为强类型的Array[Struct]，再用explode把数组拆成单行，最后提取Struct中的字段即可。全程基于Spark的分布式执行引擎，性能拉满，适合TB级数据。

具体实现步骤 #

1. 预定义JSON数组的Schema（关键优化点）

TB级数据下，不要让Spark自动推断Schema，预定义Schema能避免全表扫描推断的开销，直接指定结构：

from pyspark.sql.types import StructType, StructField, StringType, ArrayType

# 定义单个地址的Struct Schema
address_schema = StructType([
    StructField("city", StringType(), nullable=True),
    StructField("state", StringType(), nullable=True),
    StructField("street", StringType(), nullable=True),
    StructField("postalCode", StringType(), nullable=True),
    StructField("country", StringType(), nullable=True)
])

2. 解析JSON字符串并展开数组

直接用from_json解析addresses列，再用explode拆分数组，最后提取字段：

from pyspark.sql import SparkSession
from pyspark.sql.functions import from_json, explode, col

# 初始化SparkSession（根据集群配置调整参数）
spark = SparkSession.builder.appName("AddressParser").getOrCreate()

# 示例数据（实际替换为你的TB级数据源，如HDFS/S3的Parquet/ORC）
data = [
    (1, '[{"city":null,"state":null,"street":"123, ABC St, ABC  Square","postalCode":"11111","country":"USA"},{"city":"Dallas","state":"TX","street":"456, DEF Plaza, Test St","postalCode":"99999","country":"USA"}]')
]
df = spark.createDataFrame(data, ["id", "addresses"])

# 解析JSON字符串为数组类型
parsed_df = df.withColumn("addresses_array", from_json(col("addresses"), ArrayType(address_schema)))

# 展开数组为单行
exploded_df = parsed_df.withColumn("address", explode(col("addresses_array")))

# 提取Struct中的字段，生成最终结构化表格
final_df = exploded_df.select(
    col("id"),
    col("address.city"),
    col("address.state"),
    col("address.street"),
    col("address.postalCode"),
    col("address.country")
).drop("addresses", "addresses_array", "address")

# 查看结果
final_df.show(truncate=False)

3. TB级数据的性能优化建议

• 使用列式存储格式：如果源数据是CSV/JSON，先转成Parquet或ORC，Spark对列式存储的读取和解析性能提升巨大。
• 避免不必要的Shuffle：explode操作不会触发Shuffle，放心使用；后续若有聚合操作，尽量提前过滤数据。
• 调整Executor资源：根据集群规模调整spark.executor.cores、spark.executor.memory等参数，最大化并行处理能力。
• 分区优化：确保源数据的分区数合理，避免小分区或超大分区，可通过repartition或coalesce调整。

输出结果 #

执行后会得到你需要的结构化表格：

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