这个问题正如丹尼尔所说的那样 reduce 必须是结合的和交换的。以下是来源本身的原因：

val reducePartition: Iterator[T] => Option[T] = iter => {
  if (iter.hasNext) {
    Some(iter.reduceLeft(cleanF))
  } else {
    None
  }
}

请注意 reduce 在每个分区上完成的是对其迭代器的简单委托 reduceLeft . 这不会造成任何问题，因为这只是价值的积累。

val mergeResult = (index: Int, taskResult: Option[T]) => {
  if (taskResult.isDefined) {
    jobResult = jobResult match {
      case Some(value) => Some(f(value, taskResult.get))
      case None => taskResult
    }
  }
}

但是，分区的合并是个问题。在您的示例中，它是如何分解的（假设4个均匀分割的分区上有40个计数）：

A = 10; B = 10; C = 10; D = 10 //Local reductions. Everything ok
A added in = 10 //Still ok
B added in = f(10, 10) = 11 //Because your definition of f is (first + 1)
                            //This drops the second param of 10
C added in = f(11, 10) = 12 //Again, only adding 1 instead of the actual 10 count

所以，你应该更喜欢 count ，或者照丹尼尔的建议去做 map ，或者你有第三个选择 aggregate ```
rdd.aggregate(0)(_+1, +)

这将使用0作为计数的种子，在本地将1添加到累加器，然后在合并中将两个累加器添加到一起。

的文档 RDD.reduce() 说：
使用指定的交换和结合二元运算符减少此rdd的元素。 def plusOne(sum, v): return sum + 1 不是可交换的。它完全忽略其中一个参数。所以你看到的是未定义的行为(我建议考虑为什么函数必须是可交换的。如果你明白这一点，你就更了解spark了！）
解决方法是使用 RDD.count() 相反。但如果你坚持使用 reduce() ，以下是您的操作方法：

def count(rdd):
  return rdd.map(lambda x: 1).reduce(lambda a, b: a + b)

这不是完整的答案
因为我不能把我的发现发表评论，所以我把它们写在这里。
我可以用一个更简单的例子来重现你的问题。

data = xrange(1, 10000)
len(data) #output => 9999
xrangeRDD = sc.parallelize(data, 8)
print xrangeRDD.count()
def plusOne (v,sum):
  #print sum, v
  return v + 1;
a = xrangeRDD.reduce(plusOne)
print a

输出

9999
1256

为了 xrangeRDD = sc.parallelize(data, 4) 输出

9999
2502

为了 xrangeRDD = sc.parallelize(data, 1) 输出

9999
9999

因为我只改变了分区的数量，这也改变了reduce的输出，所以我认为reduce只提供了一个分区的输出，正如这里的模式所建议的那样。
我还在学习Spark的工作原理。所以我不能完全理解为什么会这样。我希望通过这个额外的细节，有人能够解释这背后的原因。