如何在已排序非键值对RDD中过滤连续重复元素行？

核心思路 #

你需要对比连续相邻的行，直接zip原RDD自身只会得到相同行的配对，完全没用。正确的做法是把原RDD和它偏移一位的版本配对，这样就能得到(第n行, 第n+1行)的相邻行对，再通过过滤保留符合要求的行。

具体实现（Scala示例） #

假设你的RDD是全局排序的，每行是一个数组（比如Array[String]），你需要对比每行的前两个元素是否相同：

1. 生成相邻行对
   先给原RDD添加索引，再通过索引关联当前行和下一行：

   val sortedRDD: RDD[Array[String]] = ... // 你的已排序RDD
   val indexedRDD = sortedRDD.zipWithIndex()
   // 将索引+1作为键，关联下一行
   val pairedRows = indexedRDD.join(indexedRDD.map { case (row, idx) => (idx + 1, row) })
   

2. 过滤并保留目标行
   遍历相邻行对，若两行的目标元素不同，则保留前一行；最后还要单独保留原RDD的最后一行（因为它没有下一行，无需过滤）：

   // 过滤出相邻行目标元素不同的前一行
   val filteredRows = pairedRows.filter { case (_, (row1, row2)) =>
     row1(0) != row2(0) || row1(1) != row2(1) // 这里替换成你要比较的元素判断逻辑
   }.map { case (_, (row1, _)) => row1 }
   
   // 获取原RDD的最后一行
   val lastRow = sortedRDD.takeOrdered(1)(Ordering.by(_.toString).reverse)
   val finalRDD = filteredRows.union(sc.parallelize(lastRow))
   

另一种轻量实现（小数据量场景） #

如果你的数据量不大，可以先将RDD合并为单个分区，然后在分区内遍历处理，这样更简单：

val finalRDD = sortedRDD.coalesce(1).mapPartitions { iter =>
  if (!iter.hasNext) iter
  else {
    var prev = iter.next()
    val result = collection.mutable.ListBuffer(prev)
    while (iter.hasNext) {
      val curr = iter.next()
      // 替换成你的判断逻辑：如果当前行和前一行的目标元素不同，才保留
      if (curr(0) != prev(0) || curr(1) != prev(1)) {
        result += curr
        prev = curr
      }
    }
    result.iterator
  }
}

注意事项 #

• 上述方法依赖RDD是全局有序的，如果你的排序只在分区内有效，需要先通过repartitionAndSortWithinPartitions实现全局有序的分区，否则跨分区的相邻行无法被正确对比。
• 不要用groupBy或reduceByKey，这类操作会把所有相同键的行聚合在一起，无法区分是否是连续的重复行，不符合你的需求。

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