Сгруппируйте по столбцам и запишите каждую группу строк в текстовый файл, используя Apache Spark и Java

Question

У меня есть файл .csv с идентификатором столбцов и парой строковых столбцов. Я хочу сгруппировать по идентификатору, а затем записать все значения из string_column1 в текстовый файл (каждое значение в новой строке). Наконец, я хочу, чтобы имя текстового файла было «allstrings» + id. Я использую Apache Spark с Java.

Я пытался использовать groupBy ("id"). Agg (collect_list ("string_column1")), но получаю "Метод collect_list (String) не определен длятип Main ".
Я не знаю, как назвать текстовые файлы, используя отличающиеся значения из столбца 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, а имя каждой папки будет иметь формат имя_ столбца = значение .

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

Затем вы можете использовать org.apache.hadoop.fs._ для переименования файлов из каждого каталога группы.