如何使用Java版Spark并行处理不同位置的多文件？求示例代码

嘿，这个问题我太熟悉了！在Java版Spark里并行处理分散在不同位置的多个文件其实非常省心——Spark的分布式架构天生就支持这种场景，而且加载多文件的方式灵活多样，我给你拆解几种常用方法，再附上完整示例代码，你一看就懂。

Spark的数据源读取API（比如textFile()、csv()、parquet()等）都支持直接传入多个路径，不管这些路径是本地文件系统还是HDFS、S3这类分布式存储。你可以把所有路径放进一个List，或者直接作为多个参数传进去。

示例代码：

import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.functions;
import java.util.Arrays;
import java.util.List;

public class MultiFileProcessing {
    public static void main(String[] args) {
        // 初始化SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("MultiFileParallelProcessing")
                .master("local[*]") // 本地模式用所有核心，生产环境请移除这行
                .getOrCreate();

        // 定义多个不同位置的文件路径
        String path1 = "/data/files/file1.txt";
        String path2 = "/data/archive/file2.txt";
        String path3 = "s3://my-bucket/data/file3.txt"; // 分布式存储路径也支持

        // 方式一：直接传入多个路径参数
        Dataset<String> textData1 = spark.read().textFile(path1, path2, path3);

        // 方式二：用List封装路径（适合路径数量多的场景）
        List<String> filePaths = Arrays.asList(path1, path2, path3);
        Dataset<String> textData2 = spark.read().textFile(filePaths.toArray(new String[0]));

        // 并行处理示例：统计每个文件的行数（通过内置函数获取文件路径）
        textData2.withColumn("file_path", functions.input_file_name())
                .groupBy("file_path")
                .count()
                .show();

        spark.stop();
    }
}

如果你的文件路径有规律（比如同前缀、同后缀，或者分散在同层级子目录里），用通配符*或者?会更高效，不用手动列所有路径。

示例代码：

// 匹配/data目录下所有子目录里的txt文件
Dataset<String> wildcardData = spark.read().textFile("/data/*/*.txt");

// 匹配以file开头、任意后缀的文件
Dataset<String> patternData = spark.read().textFile("/data/files/file*.*");

如果文件分散在多层子目录里，可以开启recursiveFileLookup选项，让Spark递归遍历所有子目录加载文件。

示例代码：

Dataset<org.apache.spark.sql.Row> csvData = spark.read()
        .option("recursiveFileLookup", "true")
        .csv("/data/root-directory/"); // 读取该目录下所有层级的csv文件

• 文件格式一致性：如果多个文件格式不同（比如有的是txt，有的是csv），你需要分别读取再合并，或者使用Spark的通用数据源API做适配处理。
• 并行度控制：Spark会根据文件数量和大小自动分配任务，你也可以通过spark.sql.shuffle.partitions配置项，或者调用repartition()方法手动调整并行度，优化处理速度。
• 存储兼容性：不管是本地文件、HDFS、S3还是Azure Blob，只要Spark配置了对应存储的依赖（比如hadoop-aws包），就能直接读取，无需额外修改代码。

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