如何在 Google Dataproc 集群上运行 Java 并行算法？

Question

我有一个使用 Spark 实现的简单 java 并行算法。但我不确定如何在 Google Dataproc 集群上运行它。我在网上找到了很多使用 python 或 scala 的资源，但对于 java 来说还不够。这是代码

public class Prime {

    List<Integer> primes = new ArrayList<>();

    //Method to calculate and count the prime numbers
    public void countPrime(int n){
        for (int i = 2; i < n; i++){
            boolean  isPrime = true;

            //check if the number is prime or not
            for (int j = 2; j < i; j++){
                if (i % j == 0){
                    isPrime = false;
                    break;  // exit the inner for loop
                }
            }

            //add the primes into the List
            if (isPrime){
                primes.add(i);
            }
        }
    }

    //Main method to run the program
    public static void main(String[]args){
        //creating javaSparkContext object
        SparkConf conf = new SparkConf().setAppName("haha").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(conf);
        //new prime object
        Prime prime = new Prime();
        prime.countPrime(100000);

        //parallelize the collection
        JavaRDD<Integer> rdd = sc.parallelize(prime.primes , 4);
        long count = rdd.filter(e  -> e == 2|| e % 2 != 0).count(); 
    }
}

Answer 1

如果您已经有一个将“Prime”指定为主类的 jarfile，那么在基本级别上它很简单：

gcloud dataproc jobs submit spark --cluster ${CLUSTER_NAME} --jar prime-jarfile.jar

如果您的 jarfile 没有指定 main-class，您可以将 jarfile 提交为“--jars”（末尾带有“s”）并指定“--class”：

gcloud dataproc jobs submit spark --cluster ${CLUSTER_NAME} --jars prime-jarfile.jar --class Prime

但是请注意，由于您指定了setMaster("local")，它会覆盖集群自己的 spark 环境设置，并且它只会使用主节点上的线程运行。您只需完全删除 .setMaster("local")，它就会自动获取 Dataproc 集群中的 YARN 配置，以便在多个工作节点上实际运行。

另外，我意识到这只是一个入门练习，所以它可能无关紧要，但您几乎肯定不会在真正的分布式模式中看到任何“加速”，因为：

1. 与加载整数所需的时间相比，使用 Spark 的计算太“便宜”了。
2. 与启动远程执行的开销相比，正在处理的元素数量太少
3. 分区数 (4) 可能太少，无法启动动态执行程序分配，因此它们最终可能会一个接一个地运行

所以你可能会看到更多“有趣”的结果，例如，如果你并行化的每个数字都代表了工作人员要检查的大“范围”；例如，如果数字“0”表示“计数 0 到 1,000,000 之间的素数”，“1”表示“计数 1,000,000 到 2,000,000 之间的素数”，等等。那么你可能会有这样的情况：

// Start with rdd is just parallelize the numbers 0 through 999 inclusive with something like 100 to 1000 "slices".
JavaRDD<Integer> countsPerRange = rdd.map(e -> countPrimesInRange(e*1000000, (e+1)*1000000));
int totalCount = countsPerRange.reduce((a, b) -> a + b);