java - 按列分组并使用 Apache Spark 和 Java 将每组字符串写入文本文件

Question

我有一个 .csv 文件，其中包含列 id 和几个字符串列。我想按 id 进行分组，然后将 string_column1 中的所有值写入文本文件(每个值都在新行上)。最后，我希望文本文件的名称为“allstrings”+id。我正在将 Apache Spark 与 Java 结合使用。

我尝试使用 groupBy("id").agg(collect_list("string_column1")) 但我得到“Main 类型的方法collect_list(String) 未定义”。
我不知道如何使用 id 列中的不同值来命名文本文件。

import org.apache.log4j.Level;
import org.apache.log4j.Logger;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.RelationalGroupedDataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

public class Main {

    public static void main(String[] args) {
        Logger.getLogger("org.apache").setLevel(Level.WARN);

        SparkSession spark = SparkSession.builder()
                .appName("testingSql")
                .master("local[*]")
                .getOrCreate();

        Dataset<Row> dataset = spark.read()
        .option("header", true)
        .csv("src/main/resources/maininput.csv");

        // make a separate .csv file for each group of strings (grouped by id),
        // with each string on a new line
        // and the name of the file should be "allstrings"+id
        RelationalGroupedDataset result = dataset.groupBy("id")
                .agg(collect_list("string_column1"))
                .?????????;



        spark.close();
    }

}

Answer 1

您可以在写入时对数据进行分区，它将为每个组id创建单独的目录每个文件夹的名称格式为column_name=value。

df.write.partitionBy("id").csv("output_directory")

然后您可以使用org.apache.hadoop.fs._重命名每个组目录中的文件。