【spark】4. Dataset转换算子使用

package com.DataSet;

import bean.Dept;
import bean.Employee;
import org.apache.spark.sql.*;

import java.util.ArrayList;
import java.util.List;


public class DataSetConvert {
    private static SparkSession spark = SparkSession.builder().master("local[*]").appName("handle data").getOrCreate();

    public static void main(String[] args) {
        spark.conf().set("spark.sql.crossJoin.enabled", "true");
        spark.sparkContext().setLogLevel("WARN");
        Encoder employeeEncoder = Encoders.bean(Employee.class);
        String path = "spark-hello/src/main/resources/employees.json";
        Dataset ds = spark.read().json(path).as(employeeEncoder);


        Dataset out = flatMapGroups(groupByKey(ds));
        out.show();

        Dataset out2 = mapGroups(groupByKey(ds));
        out2.show();


    }

    public static KeyValueGroupedDataset groupByKey(Dataset ds) {
        return ds.groupByKey(e -> e.getName(), Encoders.STRING());
    }

    public static Dataset mapGroups(KeyValueGroupedDataset kvgDS) {
        Dataset out = kvgDS.mapGroups((key, eList) -> {
            Dept dept = new Dept();
            eList.forEachRemaining(e -> {
                dept.addEmployee(e);
            });
            return dept;
        }, Encoders.bean(Dept.class));
        return out;
    }

    public static Dataset flatMapGroups(KeyValueGroupedDataset kvgDS) {
        Dataset out = kvgDS.flatMapGroups((key, eList) -> {
            List employees = new ArrayList<>();
            eList.forEachRemaining(e -> {
                employees.add(e);
            });
            return employees.iterator();

        }, Encoders.bean(Employee.class));
        return out;
    }


}