基于Spark的大数据分析代码运行出现AssertionError如何解决？

报错根因 #

报错触发于assert counts3[0][0] == 5.0行，和Spark版本无关，属于逻辑设计问题：Spark的reduceByKey是Shuffle类操作，输出结果的顺序没有固定保证，collect()返回的列表中第一个元素的label不一定是预期的5.0，直接按下标取数断言就会触发报错。

可能诱因 #

• Shuffle操作输出的RDD分区内、分区之间的元素顺序默认是随机的，没有额外排序逻辑的前提下，collect()返回的列表顺序不固定，无法保证第一个元素的label为5.0
• 采样的3条数据Data.take(3)的label分布不符合预期：要么3条数据中label为5.0的条目数不是2，要么根本没有label为5.0的条目
• 若Data对应的RDD没有做持久化，每次执行take、count这类Action操作都会重新读取数据源，可能出现多次采样拿到的数据不一致的问题

修复方案 #

1. 调整计数方法，对分组后的结果按label排序后再返回，保证返回列表的顺序固定：

def count_examples(Data):
    count = Data.map(lambda labelPoint: (labelPoint.label,1)).reduceByKey(lambda x,y: x+y)
    # 按label升序排序，固定返回结果的顺序
    return count.sortByKey().collect()

2. 校验采样数据的label分布是否符合预期，执行以下代码打印采样的所有label值：

sample_labels = [lp.label for lp in Data.take(3)]
print(sample_labels)

如果输出结果中5.0的数量不是2，调整你的采样逻辑即可。
3. 若你的Data是从动态数据源加载生成的，可先执行Data.persist()将RDD持久化到内存，避免多次Action操作读取到不同的数据。

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