如何在Apache Spark中并行加载DataFrame以优化作业效率？

我需要创建两个包含不同数据的DataFrame，希望并行加载对应的CSV文件。
算法步骤：

1. 加载X DataFrame，执行操作构建聚合视图并转换为Y DataFrame（该步骤耗时约30分钟）
2. 加载Z DataFrame，并将其与Y DataFrame进行关联。

为节省时间，我希望在执行步骤1的同时并行加载Z DataFrame，以便Y准备好后可直接进行关联及后续操作。请问如何实现最优方案？我想到的一种方法是使用线程，但这是否是最佳方案？

我写的代码示例： #

// 使用线程实现并行化
Thread thread1 = new Thread(() -> {
    Dataset<Row> df1 = spark.read()
            .option("multiLine", true)  // 处理多行CSV记录（如果适用）
            .csv(folder1Path);
    // 在这里对df1执行必要的转换操作
});

Thread thread2 = new Thread(() -> {
    Dataset<Row> df2 = spark.read()
            .option("multiLine", true)  // 处理多行CSV记录（如果适用）
            .csv(folder2Path);
    // 在这里对df2执行必要的转换操作
});

// 同时启动线程
thread1.start();
thread2.start();

// 等待线程执行完成
try {
    thread1.join();
    thread2.join();
} catch (InterruptedException e) {
    e.printStackTrace();
    System.exit(1);
}

直接用Java线程的问题 #

你想到的线程方法不是Spark场景下的最优方案，核心问题有两个：

1. Spark的SparkSession并非线程安全，多线程中直接调用spark.read()可能引发资源竞争、状态混乱等并发问题，甚至导致任务失败。
2. Spark本身是分布式调度框架，手动创建客户端线程不会让集群资源得到更高效的利用，反而可能因为客户端线程竞争干扰Spark的任务调度逻辑。

最优实现方案：Spark异步API + CompletableFuture #

结合Spark的异步作业提交能力与Java的CompletableFuture，可以安全实现并行处理，同时适配Spark的调度体系。具体逻辑如下：

1. 用CompletableFuture异步执行Y DataFrame的构建流程（步骤1）
2. 同时启动另一个CompletableFuture异步加载Z DataFrame（步骤2）
3. 等待两个任务完成后，执行关联操作

优化后的代码示例 #

import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import org.apache.spark.sql.functions;

// 异步构建Y DataFrame（对应步骤1的聚合转换）
CompletableFuture<Dataset<Row>> yDfFuture = CompletableFuture.supplyAsync(() -> {
    Dataset<Row> xDf = spark.read()
            .option("multiLine", true)
            .csv(folder1Path);
    // 示例聚合操作，替换为你的实际转换逻辑
    return xDf.groupBy("target_col").agg(functions.sum("metric").as("total_metric"));
}, spark.sparkContext().environment().executorService());

// 异步加载Z DataFrame（对应步骤2）
CompletableFuture<Dataset<Row>> zDfFuture = CompletableFuture.supplyAsync(() -> {
    return spark.read()
            .option("multiLine", true)
            .csv(folder2Path);
}, spark.sparkContext().environment().executorService());

// 等待两个任务完成，执行关联
try {
    Dataset<Row> yDf = yDfFuture.get();
    Dataset<Row> zDf = zDfFuture.get();
    
    // 执行关联操作，替换为你的实际关联逻辑
    Dataset<Row> joinedDf = yDf.join(zDf, yDf.col("id").equalTo(zDf.col("y_ref_id")), "inner");
    // 后续处理逻辑...
} catch (InterruptedException | ExecutionException e) {
    e.printStackTrace();
    System.exit(1);
}

关键注意事项 #

• 使用Spark自带的executorService提交异步任务，确保与Spark的调度体系兼容，避免线程安全问题。
• Spark的DataFrame转换是懒执行的，supplyAsync中仅定义执行计划，实际计算会在调用get()时触发，确保两个任务能在集群上并行执行。
• 对于耗时较长的步骤1，并行加载Z的操作能充分利用集群空闲资源，有效缩短整体任务耗时。

内容的提问来源于stack exchange，提问作者Smit