java实现spark读取hdfs目录下5个文件,给每个文件中的数据单独加行号

在Java中使用Apache Spark读取HDFS目录下的多个文件并对每个文件的数据添加行号，你可以按照以下步骤操作：

首先，确保已经安装了Spark和相关的库，如Hadoop。然后，可以使用`SparkSession`来创建一个Spark环境，并使用`DataFrameReader`从HDFS读取文件。这里假设你已经设置了Hadoop配置。

import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class FileLineNumberAdder {
    public static void main(String[] args) {
        // 创建SparkSession
        SparkSession spark = SparkSession.builder()
                .appName("HDFSFileLineNumberAdder")
                .config("spark.hadoop.fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem")
                .getOrCreate();

        String hdfsPath = "hdfs://your_hdfs_path"; // 替换为你的HDFS路径
        List<String> filesToProcess = Arrays.asList("file1.txt", "file2.txt", "file3.txt", "file4.txt", "file5.txt");

        for (String file : filesToProcess) {
            Dataset<Row> df = spark.read()
                    .option("header", "true") // 如果文件有列名，设置为true
                    .csv(hdfsPath + "/" + file);

            // 对每一行添加行号
            Dataset<Row> numberedDF = df.selectExpr("_c0 as data", "(row_number() over(order by _c0)) as line_num");

            // 打印结果或者保存到新的文件
            numberedDF.show();
            // 可能的话，将处理后的DataFrame保存到一个新的文件，例如 "output/" + file + "_numbered.csv"
            numberedDF.write().format("csv").save("output/" + file + "_numbered.csv");
        }

        spark.stop(); // 关闭SparkSession
    }
}

在这个示例中，我们遍历文件列表，对每一份文件读取数据，然后用`row_number()`函数给每行添加行号。最后，你可以选择打印出来查看，或者保存成新的CSV文件。