PySpark：如何从邮编数组列提取区域并找出最常见值？

步骤分解与代码实现 #

假设你的原始DataFrame包含postcode_outcode数组列和其他需要保留的列，我们可以通过以下步骤实现需求：

1. 初始化环境并构造示例DataFrame

from pyspark.sql import SparkSession
from pyspark.sql import functions as F
from pyspark.sql.window import Window

spark = SparkSession.builder.appName("PostcodeAreaExtraction").getOrCreate()

# 模拟原始数据，包含需要保留的其他列
data = [
    (["SN00", "SN01", "OA77"], "value1"),
    (["FG12", "PC00", "PD01", "FG99"], "value2"),
    (["WP55"], "value3"),
    (["DD01", "DD02"], "value4")
]
df = spark.createDataFrame(data, ["postcode_outcode", "other_column"])

2. 给每行添加唯一标识

为了炸分数组后能对应回原始行，先给每行生成唯一ID：

df_with_id = df.withColumn("row_id", F.monotonically_increasing_id())

3. 炸分数组列

将数组中的每个邮编外码拆分为单独一行：

exploded_df = df_with_id.select(
    "row_id", "postcode_outcode", "other_column",
    F.explode("postcode_outcode").alias("single_outcode")
)

4. 提取邮编区域

用正则表达式提取每个外码开头的字母部分（即邮编区域）：

area_extracted_df = exploded_df.withColumn(
    "postcode_area",
    F.regexp_extract("single_outcode", "^[A-Za-z]+", 0)
)

5. 统计每行内各区域的出现次数

按行ID和邮编区域分组，统计每个区域在当前行的出现次数：

area_count_df = area_extracted_df.groupBy(
    "row_id", "postcode_area"
).agg(F.count("*").alias("area_occurrence"))

6. 筛选每行最常见的邮编区域

使用窗口函数按行ID分区，按出现次数降序排序，取每个分区的第一行（即出现次数最多的区域）：

window_spec = Window.partitionBy("row_id").orderBy(F.desc("area_occurrence"))
most_common_area_df = area_count_df.withColumn(
    "rank", F.row_number().over(window_spec)
).filter(F.col("rank") == 1).select("row_id", "postcode_area")

7. 关联回原始DataFrame，保留所有列

将筛选出的最常见区域和原始DataFrame关联，去掉临时的row_id，得到最终结果：

final_result = df_with_id.join(most_common_area_df, on="row_id", how="inner") \
    .drop("row_id") \
    .select("other_column", "postcode_outcode", "postcode_area")

final_result.show()

补充说明 #

• 如果同一行内多个区域出现次数相同（比如一行里SN和OA各出现1次），row_number()会随机取其中一个；若想保留所有并列最多的区域，可以改用rank()并去掉filter(F.col("rank") == 1)的限制。
• 正则表达式^[A-Za-z]+适配英国邮编外码的区域规则，确保只提取开头的字母部分。

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