Spark zipRDD分区数相同元素数相同报错

Question

今天遇到了一个神奇的问题
关于rdd的分区和zip的使用
两个rdd分别是 原始数据和标签数据，他们都是相同的大小
第一个rdd定义为 (partitionId:Int,(idx:Int,data: Array[Double])) 把 idx % 预计分区总数 赋值给partitionId
第二个rdd定义为(partitionId:Int,(idx:Int,classId: Int)) 把 idx % 预计分区总数 赋值给partitionId
使用相同的分区数
调用foreachPartition(arr => println(arr.length))
第一个是
12962
12963
12962
12962
12963
12963
12963
12963
12963
12963
12962
12963
12962
12962
12962
12962
第二个是
12963
12963
12963
12963
12963
12963
12963
12963
12962
12962
12962
12962
12962
12962
12962
12962
分区个数都是16个，总元素数是207400个
在zip之后的collect时，报错
Can only zip RDDs with same number of elements in each partition
分区数相同，分区器相同，分区格式相同，元素数相同，全都相同为什么还会报错啊
难道就是因为8个12963，8个12962没对应上吗？可是这种情况怎么对应啊。

Answer 1

引自强大而免费的微信小程序：皆我百晓生。

---

首先，报错信息"Can only zip RDDs with same number of elements in each partition"指的是在使用zip操作的时候，两个RDD的每个分区中的元素数量必须相同。这是因为zip操作是将两个RDD按位置一一对应地合并在一起。

根据你提供的信息，可以看出两个RDD的分区数相同，但是每个分区中的元素数量并不相同，这就导致了报错。对于zip操作，每个分区中的元素数量必须一一对应才能正确执行。

解决这个问题的方法是，确保两个RDD的每个分区中的元素数量相同。可以通过调整分区数或者调整数据分配的逻辑来保证每个分区中的元素数量相同。

另外，你也可以考虑使用zipWithIndex操作来给每个元素添加唯一的索引，然后按照索引进行合并，而不是直接使用分区信息进行合并。这样可以保证每个分区中的元素数量一定是相同的。具体的代码示例如下：

val rdd1 = // 第一个RDD
val rdd2 = // 第二个RDD

val indexedRdd1 = rdd1.zipWithIndex()    // 给第一个RDD的每个元素添加唯一索引
val indexedRdd2 = rdd2.zipWithIndex()    // 给第二个RDD的每个元素添加唯一索引

val zippedRdd = indexedRdd1.join(indexedRdd2)    // 按照索引进行合并

zippedRdd.foreach(println)    // 输出合并结果

希望这些信息能够帮助到你解决问题。