在 Apache Spark 中，我可以轻松 repeat/nest 一个 SparkContext.parallelize 吗？

Question

我正在尝试对我们正在尝试解决的遗传学问题进行建模，逐步构建它。我可以成功运行来自 Spark Examples 的 PiAverage 示例。该示例 "throws darts" 在一个圆圈（在我们的例子中为 10^6）并计算 "land in the circle" 的数量以估计 PI

假设我想重复该过程 1000 次（并行）并对所有这些估计进行平均。我想看看最好的方法，似乎会有两个并行化调用？嵌套调用？有没有办法将 map 或 reduce 调用链接在一起？我看不到。

我想知道像下面这样的想法的智慧。我想到了使用累加器来跟踪估算结果。 jsc 是我的 SparkContext，单个运行的完整代码在问题的最后，感谢您的任何输入！

Accumulator<Double> accum = jsc.accumulator(0.0);

// make a list 1000 long to pass to parallelize (no for loops in Spark, right?)
List<Integer> numberOfEstimates = new ArrayList<Integer>(HOW_MANY_ESTIMATES);

// pass this "dummy list" to parallelize, which then 
// calls a pieceOfPI method to produce each individual estimate  
// accumulating the estimates. PieceOfPI would contain a 
// parallelize call too with the individual test in the code at the end
jsc.parallelize(numberOfEstimates).foreach(accum.add(pieceOfPI(jsc, numList, slices, HOW_MANY_ESTIMATES)));

// get the value of the total of PI estimates and print their average
double totalPi = accum.value();

// output the average of averages
System.out.println("The average of " + HOW_MANY_ESTIMATES + " estimates of Pi is " + totalPi / HOW_MANY_ESTIMATES);

它看起来不像我在 SO 上看到的矩阵或其他答案，所以给出了这个特定问题的答案，我已经做了几次搜索，但我不知道如何在没有 "parallelizing the parallelization." 的情况下做到这一点是吗一个坏主意？

（是的，我意识到在数学上我可以做更多的估计并有效地得到相同的结果:)试图建立我老板想要的结构，再次感谢！

如果有帮助的话，我已经把我的整个单一测试程序放在这里了，没有我正在测试的累加器。这个的核心将成为 PieceOfPI():

import java.io.Serializable;
import java.util.ArrayList;
import java.util.List;

import org.apache.spark.Accumulable;
import org.apache.spark.Accumulator;
import org.apache.spark.SparkContext;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.storage.StorageLevel;
import org.apache.spark.SparkConf;
import org.apache.spark.storage.StorageLevel;

public class PiAverage implements Serializable {

public static void main(String[] args) {

    PiAverage pa = new PiAverage();
    pa.go();

}

public void go() {

    // should make a parameter like all these finals should be
    // int slices = (args.length == 1) ? Integer.parseInt(args[0]) : 2;
    final int SLICES = 16;

    // how many "darts" are thrown at the circle to get one single Pi estimate
    final int HOW_MANY_DARTS = 1000000;

    // how many "dartboards" to collect to average the Pi estimate, which we hope converges on the real Pi
    final int HOW_MANY_ESTIMATES = 1000;

    SparkConf sparkConf = new SparkConf().setAppName("PiAverage")
        .setMaster("local[4]");

    JavaSparkContext jsc = new JavaSparkContext(sparkConf);

    // setup "dummy" ArrayList of size HOW_MANY_DARTS -- how many darts to throw
    List<Integer> throwsList = new ArrayList<Integer>(HOW_MANY_DARTS);
    for (int i = 0; i < HOW_MANY_DARTS; i++) {
        throwsList.add(i);
    }

    // setup "dummy" ArrayList of size HOW_MANY_ESTIMATES
    List<Integer> numberOfEstimates = new ArrayList<Integer>(HOW_MANY_ESTIMATES);
    for (int i = 0; i < HOW_MANY_ESTIMATES; i++) {
        numberOfEstimates.add(i);
    }

    JavaRDD<Integer> dataSet = jsc.parallelize(throwsList, SLICES);

    long totalPi = dataSet.filter(new Function<Integer, Boolean>() {
        public Boolean call(Integer i) {
            double x = Math.random();
            double y = Math.random();
            if (x * x + y * y < 1) {
                return true;
            } else
                return false;
        }
    }).count();

    System.out.println(
            "The average of " + HOW_MANY_DARTS + " estimates of Pi is " + 4 * totalPi / (double)HOW_MANY_DARTS);

    jsc.stop();
    jsc.close();
}
}

Answer 1

让我从你的 "background question" 开始。 map、join、groupBy等变换操作分为两类；那些需要将数据洗牌作为来自所有分区的输入的那些，以及那些不需要的。 groupBy 和 join 之类的操作需要洗牌，因为您需要使用相同的键将来自所有 RDD 分区的所有记录放在一起（想想 SQL JOIN 和 GROUP BY 操作工作）。另一方面，map、flatMap、filter等不需要洗牌，因为操作在上一步分区的输入上工作正常。他们一次处理单个记录，而不是具有匹配键的一组记录。因此，不需要洗牌。

这个背景对于理解 "extra map" 没有显着的开销是必要的。 map、flatMap 等一系列操作 "squashed" 一起构成 "stage"（当您在 Spark Web 控制台中查看作业的详细信息时会显示) 这样只有一个 RDD 被具体化，即阶段末尾的那个。

关于你的第一个问题。我不会为此使用累加器。它们用于 "side-band" 数据，例如计算您解析了多少错误行。在此示例中，您可以使用累加器来计算有多少 (x,y) 对在 1 的半径内与外部，例如。

Spark 发行版中的 JavaPiSpark 示例已经差不多好了。你应该研究它为什么有效。它是适用于大数据系统的正确数据流模型。您可以使用 "aggregators"。在 Javadocs 中，单击 "index" 并查看 agg、aggregate 和 aggregateByKey 函数。但是，它们在这里不再是可以理解的，也不是必需的。它们比 map 和 reduce 提供更大的灵活性，因此值得了解

您的代码的问题在于，您实际上是在尝试告诉 Spark 做什么，而不是表达您的意图并让 Spark 优化它为您做事的方式。

最后，建议大家购买学习O'Reilly的"Learning Spark"。它很好地解释了内部细节，例如暂存，并且它显示了许多您也可以使用的示例代码。

在 Apache Spark 中，我可以轻松 repeat/nest 一个 SparkContext.parallelize 吗？

In Apache Spark, can I easily repeat/nest a SparkContext.parallelize?

java

parallel-processing

scala

nested-loops

apache-spark