PySpark DataFrame自定义分组：基于相似地址构建家庭数据集

这是个典型的**实体解析（Entity Resolution）**场景——没有统一的主键，得靠自定义相似性规则来识别关联记录。我给你一套实操性强的方案，用GraphFrames处理连通分组，这是目前最直接高效的方式：

步骤1：准备环境与注册地址匹配UDF #

首先确保你已引入GraphFrames依赖（Maven/Gradle添加对应依赖即可），然后把你的自定义地址匹配逻辑封装成Spark UDF：

// 示例自定义匹配函数（你可以替换成自己的判断逻辑）
def isSameAddress(address1: String, address2: String): Boolean = {
  // 先标准化地址：统一大小写、替换缩写、去除冗余表述
  val normalize = (addr: String) => addr.toLowerCase()
    .replaceAll("st\\.|str", "street")
    .replaceAll("h\\.no|house", "house number")
    .replaceAll("\\s+", " ") // 合并多余空格

  normalize(address1) == normalize(address2)
}

// 注册成Spark UDF
val isSameAddressUdf = org.apache.spark.sql.functions.udf(isSameAddress _)

步骤2：生成匹配的地址对 #

将原DataFrame自连接，过滤出符合匹配规则的记录对（只保留序号a < 序号b的对，避免重复计算）：

import org.apache.spark.sql.functions.col

// 初始化你的原始DataFrame
val rawDf = spark.createDataFrame(Seq(
  (1, "st. 1 h.no 16", "X", "1-1-2001"),
  (2, "str n.1 house 16", "Y", "1-5-2001"),
  (3, "st. 3 h.no 1", "Z", "1-8-2002")
)).toDF("序号", "地址", "姓名", "出生日期")

// 自连接并过滤匹配对
val matchedPairs = rawDf.alias("a")
  .join(rawDf.alias("b"), col("a.序号") < col("b.序号"))
  .filter(isSameAddressUdf(col("a.地址"), col("b.地址")))
  .select(col("a.序号").alias("src"), col("b.序号").alias("dst"))

步骤3：用GraphFrames计算连通分量 #

把每条记录当作图的节点，匹配的地址对当作节点间的边，通过连通分量算法找到属于同一地址的所有记录：

import org.graphframes.GraphFrame

// 构建图的节点表（用序号作为节点ID）
val vertices = rawDf.select(col("序号").alias("id"))

// 构建图的边表（来自之前的匹配对）
val edges = matchedPairs

// 初始化图并计算连通分量
val graph = GraphFrame(vertices, edges)
val connectedComponents = graph.connectedComponents.run()

步骤4：合并结果并构建家庭DataFrame #

把连通组件ID关联回原表，再按组件ID分组聚合，得到最终的家庭分组数据：

// 关联组件ID到原始数据
val dfWithGroup = rawDf.join(connectedComponents, rawDf("序号") === connectedComponents("id"))
  .drop("id")

// 按组件ID分组，生成家庭DataFrame
val familyDf = dfWithGroup.groupBy("component")
  .agg(
    org.apache.spark.sql.functions.collect_list("姓名").alias("家庭成员"),
    org.apache.spark.sql.functions.collect_list(org.apache.spark.sql.functions.struct("序号", "姓名", "出生日期")).alias("成员详情"),
    org.apache.spark.sql.functions.first("地址").alias("参考地址") // 可选：用第一条地址作为组内参考，或自行标准化
  )

// 查看结果
familyDf.show(false)

替代方案：无GraphFrames时的迭代分组 #

如果你的环境无法引入GraphFrames，也可以用迭代式标记分组（适合小数据量）：

1. 给每条记录初始组ID为自身序号
2. 迭代查找匹配记录，将组ID更新为组内最小序号
3. 直到没有组ID更新为止

不过这种方法在数据量大时性能很差，优先推荐GraphFrames方案。

关键注意事项 #

• 地址标准化优先：在匹配前先统一地址格式（比如替换缩写、大小写、去除冗余词），能大幅提升匹配准确性
• 减少计算量：大数据量下，先按地址前缀、邮编等做预分组，再在组内做匹配，避免全量自连接的性能问题
• Python环境适配：逻辑完全一致，只需把Scala语法换成Python，UDF用pyspark.sql.functions.udf即可

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