在 apache spark 中运行的作业超出预期

Question

我正在尝试学习 apache-spark。这是我正在尝试运行的代码。我正在使用pyspark api。

data = xrange(1, 10000)
xrangeRDD = sc.parallelize(data, 8)
def ten(value):
  """Return whether value is below ten.

  Args:
      value (int): A number.

  Returns:
      bool: Whether `value` is less than ten.
  """
  if (value < 10):
      return True
  else:
      return False
filtered = xrangeRDD.filter(ten)

print filtered.collect()
print filtered.take(8)

print filtered.collect() 将此作为输出[1, 2, 3, 4, 5, 6, 7, 8, 9]。

据我了解，filtered.take(n) 将从 RDD 中获取 n 个元素并打印出来。

我正在尝试两种情况：-

1)给出的n值小于或等于RDD中的元素个数

2)n 的值大于 RDD 中的元素个数

我有pyspark application UI 可以查看每种情况下运行的作业数量。在第一种情况下，只有one job 正在运行，但在第二种情况下，five jobs 正在运行。

我无法理解为什么会这样。提前致谢。

Answer 1

RDD.take 尝试评估尽可能少的分区。

如果你take(9) 它将获取分区 0（作业 1）找到 9 个项目并愉快地终止。

如果您take(10)，它将获取分区 0（作业 1）并找到 9 个项目。它还需要一个。由于分区 0 有 9 个，它认为分区 1 可能至少还有一个（作业 2）。但事实并非如此！在 2 个分区中，它找到了 9 个项目。到目前为止，每个分区有 4.5 个项目。该公式将其除以 1.5 表示悲观，并决定 10 / (4.5 / 1.5) = 3 个分区将执行此操作。所以它获取分区 2（作业 3）。依然没有。所以到目前为止每个分区有 3 个项目，除以 1.5 意味着我们需要 10 / (3 / 1.5) = 5 个分区。它获取分区 3 和 4（作业 4）。没有。我们每个分区有 1.8 个项目，10 / (1.8 / 1.5) = 8。它获取最后 3 个分区（作业 5），仅此而已。

此算法的代码在RDD.scala 中。正如你所看到的，它只不过是启发式方法。它通常会节省一些工作，但在退化的情况下会导致不必要的工作。