SparkSQL关联分组后出现重复行问题排查及疑问

首先得明确：groupBy("hid", "pid")本身就是按这两个字段的唯一组合来聚合的，理论上输出结果里(hid, pid)的组合应该是唯一的。测试环境结果正常但生产环境出现重复，大概率是生产数据存在脏数据，而非Spark的groupBy逻辑问题，下面具体拆解：

核心原因：生产数据的hid/pid存在隐形差异 #

测试数据是人工构造的干净数据，但真实生产数据可能存在以下情况，导致看起来相同的hid/pid实际是不同的分组键：

1. 隐形字符：比如字段前后有空格、换行符、不可见ASCII字符（如\t、\u0000），肉眼无法区分，但Spark会把它们当作不同的值处理。
2. 大小写不一致：比如Hid1和hid1、PID2和pid2，在Spark中是不同的字符串，会被分成不同的组。
3. Null值：如果hid或pid存在Null值，Spark会把所有Null值归为一个组，但如果后续操作中你误以为这是"重复行"，也会产生误解。

关于.drop("uid")丢失行的问题 #

drop("uid")本身不会导致数据丢失——因为join后的数据集里，uid只是一个普通字段，删除它不会减少行数。你看到的"丢失行"可能是：

• 分组后某些(hid, pid)组合的聚合结果被你忽略了（比如Null值组）；
• 生产数据中存在uid匹配但hid/pid为Null的行，这些行在分组后会单独成组，如果你没注意到Null值的组，就会误以为行丢失。

解决步骤与代码示例 #

1. 先排查脏数据 #

可以先运行以下代码，检查生产数据中hid/pid的异常情况：

// 检查hid是否有前后空格
df1.select("hid", length(hid), length(trim(hid))).where(length(hid) != length(trim(hid))).show()
// 检查pid是否有大小写差异
df2.select("pid", lower(pid)).distinct().groupBy(lower(pid)).count().where(count > 1).show()
// 检查是否有Null值
df1.join(df2, "uid").where(hid.isNull or pid.isNull).count()

2. 清洗数据后再聚合 #

针对上述脏数据问题，先对hid/pid做清洗，再执行groupBy：

import org.apache.spark.sql.functions._

df1.drop("sv")
  .join(df2, "uid")
  // 清洗：去除前后空格、统一为小写、替换隐形字符
  .withColumn("clean_hid", regexp_replace(lower(trim(col("hid"))), "\\s+", ""))
  .withColumn("clean_pid", regexp_replace(lower(trim(col("pid"))), "\\s+", ""))
  // 用清洗后的字段分组
  .groupBy("clean_hid", "clean_pid")
  .agg(
    count("*").alias("xcnt"),
    sum("sv").alias("xsum"),
    avg("sv").alias("xavg")
  )
  // 还原字段名（可选）
  .select(
    col("clean_hid").alias("hid"),
    col("clean_pid").alias("pid"),
    "xcnt", "xsum", "xavg"
  )
  .orderBy("hid")
  .show()

3. 关于重分区的疑问 #

重分区（repartition）不会解决分组重复的问题——它只是调整数据的分区数量，优化shuffle的并行度，无法处理数据本身的脏数据问题。只有先清洗数据，才能从根本上解决重复行的问题。

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