Spark 并行运行多个操作

Question

我有一个 Spark 应用程序选择一个子集并对该子集执行一些操作。每个子集与其操作之间没有依赖和交互，因此我尝试使用多线程让它们并行运行以提高性能。代码如下：

Dataset<Row> fullData = sparkSession.read().json("some_path");
ExecutorService executor = Executors.newFixedThreadPool(10);
List<Runnable> tasks = Lists.newArrayList();
for (int i = 1; i <= 50; i++) {
    final int x = i;
    tasks.add(() -> {
        Dataset<Row> subset_1 = fullData.filter(length(col("name")).equalTo(x));
        Dataset<Row> subset_2 = fullData.filter(length(col("name")).equalTo(x));
        Dataset<Row> result = subset_1.join(subset_2, ...);
        log.info("Res size is " + result.count()); // force Spark do the join operation
    });
}
CompletableFuture<?>[] futures = tasks.stream()
    .map(task -> CompletableFuture.runAsync(task, executor))
    .toArray(CompletableFuture[]::new);
CompletableFuture.allOf(futures).join();
executor.shutdown();

在 Spark 作业管理 UI 中，我注意到这 50 个任务是并行提交的，但处理仍然是阻塞方式，一个任务开始运行，直到另一个任务完成。如何让多个任务并行运行，而不是一个接一个？

Answer 1

这不是您在 Spark 中控制并行性的方式。这一切都是通过配置以声明方式控制的。

Spark 是一个分布式计算框架，它旨在用于分布式环境中，其中每个工作器都以单线程方式运行。通常任务是使用 Yarn 调度的，它具有节点的元数据，并且可能会在单个节点上启动多个任务（取决于内存和 cpu 限制）但在单独的 jvm 中。

在local 模式下，您可以将多个worker 实现为单独的线程，因此如果您说master("local[8]")，您将获得8 个worker，每个worker 在单个jvm 中作为一个线程运行。

您是如何运行您的应用程序的？