Sparklyr按列分区DataFrame异常：同列值未完全归至同一分区

原因分析 #

你遇到的问题核心是：通过sdf_repartition按origin_dest分区时，无法保证同一origin_dest的所有数据严格落在同一个分区内，导致后续spark_apply每个分区独立计算时，同一个origin_dest被多次统计，最终结果出现重复行。

具体原因：

• sdf_repartition(partition_by = ...) 依赖Spark的哈希分区策略：对origin_dest计算哈希值后，通过取模分配到对应分区。即使分区数设置为唯一起讫对数量，也可能因哈希碰撞（不同键哈希值相同）或集群内部数据分配逻辑，导致同一origin_dest的数据被拆分到多个分区。
• 你的自定义函数在每个分区内对origin_dest做分组聚合，若同一origin_dest跨多个分区，每个分区都会生成一条该路线的统计结果，最终合并后就会出现重复行，导致结果行数多于预期的4742。

另外需排查：生成origin_dest时是否存在隐形字符差异（如前后空格、大小写不一致），这种情况会让你统计的唯一起讫对数量小于实际数据中的唯一值数量，也会导致结果行数增加。

解决方案 #

方案1：使用spark_apply的group_by参数（推荐） #

spark_apply本身支持group_by参数，可直接指定按origin_dest分组，Spark会自动将同一组的所有数据分配到同一个分区，再对每个组执行自定义函数，无需手动分区：

average_by_route <- function(df) {
  library(dplyr)
  df %>%
    summarize(
      AVG_ARR_DELAY = mean(ARR_DELAY, na.rm = TRUE),
      AVG_DEP_DELAY = mean(DEP_DELAY, na.rm = TRUE)
    )
}

result <- flight_sdf %>%
  spark_apply(average_by_route, 
              group_by = "origin_dest", 
              packages = c("dplyr"))

sdf_nrow(result)

注意：此时自定义函数内无需再group_by(origin_dest)，因为每个传入函数的df已经是单个origin_dest的所有数据。

方案2：使用sdf_map_groups（贴合dplyr风格） #

sparklyr提供sdf_map_groups函数，专门用于对分组后的DataFrame应用自定义函数，逻辑更清晰：

average_by_route <- function(df) {
  df %>%
    summarize(
      AVG_ARR_DELAY = mean(ARR_DELAY, na.rm = TRUE),
      AVG_DEP_DELAY = mean(DEP_DELAY, na.rm = TRUE)
    )
}

result <- flight_sdf %>%
  group_by(origin_dest) %>%
  sdf_map_groups(average_by_route)

sdf_nrow(result)

方案3：手动保证分区与唯一键严格对应 #

如果必须手动控制分区，可以先为每个origin_dest分配唯一索引，再按索引分区，确保每个唯一值对应一个分区：

# 为每个origin_dest分配唯一索引
flight_sdf <- flight_sdf %>%
  sdf_with_unique_id("temp_id") %>%
  group_by(origin_dest) %>%
  mutate(partition_idx = first(temp_id)) %>%
  ungroup()

# 按索引分区，分区数设置为唯一起讫对数量
result <- flight_sdf %>%
  sdf_repartition(num_origin_dest, partition_by = "partition_idx") %>%
  spark_apply(average_by_route, packages = c("dplyr")) %>%
  distinct(origin_dest, .keep_all = TRUE) # 最终去重兜底

额外排查步骤 #

验证origin_dest是否存在隐形差异：

# 检查是否有前后空格
flight_sdf %>%
  mutate(origin_dest_trimmed = trimws(origin_dest)) %>%
  filter(origin_dest != origin_dest_trimmed) %>%
  sdf_nrow()

# 检查大小写差异（如果是字符串类型）
flight_sdf %>%
  mutate(origin_dest_lower = tolower(origin_dest)) %>%
  group_by(origin_dest_lower) %>%
  filter(n_distinct(origin_dest) > 1) %>%
  sdf_nrow()

如果存在这类差异，先清理数据：

flight_sdf <- flight_sdf %>%
  mutate(origin_dest = trimws(tolower(paste0(ORIGIN_AIRPORT_ID,"_", DEST_AIRPORT_ID))))

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