关于springboot:揭开神秘面纱会stream流就会大数据

如果你会任意一门语言的stream流，没道理不会大数据开发。
俗话说男追女隔座山，女追男隔层纱。
如果说零根底学大数据，感觉后面是一座山，那么只有你会java或者任意一门语言的stream流，那大数据就只隔了一层纱。
本文以java stream流计算为例，解说一些根底的spark操作。另一个风行的大数据框架flink同理。

筹备工作
测试数据,以下列别离示意姓名，年龄，部门，职位。
复制代码张三,20,研发部,普通员工
李四,31,研发部,普通员工
李丽,36,财务部,普通员工
张伟,38,研发部,经理
杜航,25,人事部,普通员工
周歌,28,研发部,普通员工

创立一个Employee类。
less复制代码 @Getter

@Setter
@AllArgsConstructor
@NoArgsConstructor
@ToString
static
class Employee implements Serializable {
    private String name;
    private Integer age;
    private String department;
    private String level;
}

}

版本：
jdk:1.8
spark:3.2.0
scala:2.12.15
下面的scala版本只是spark框架自身须要依赖到scala。
因为scala的确是比拟小众的语言，本文还是应用java演示spark代码。
1.map类
1.1 java stream map
map示意一对一操作。将上游数据的一行数据进行任意操作，最终失去操作后的一条数据。
这种思维，在java和spark,flink都是统一的。
咱们先用java stream演示读取文件，再应用map操作将每行数据映射为Employee对象。
ini复制代码List<String> list = FileUtils.readLines(new File(“f:/test.txt”), “utf-8”);

    List<Employee> employeeList = list.stream().map(word -> {
        List<String> words = Arrays.stream(word.split(",")).collect(Collectors.toList());
        Employee employee = new Employee(words.get(0), Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        return employee;
    }).collect(Collectors.toList());

    employeeList.forEach(System.out::println);

转换后的数据：
ini复制代码JavaStreamDemo.Employee(name=张三, age=20, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=李丽, age=36, department=财务部, level=普通员工)
JavaStreamDemo.Employee(name=张伟, age=38, department=研发部, level=经理)
JavaStreamDemo.Employee(name=杜航, age=25, department=人事部, level=普通员工)
JavaStreamDemo.Employee(name=周歌, age=28, department=研发部, level=普通员工)

1.2 spark map
首先失去一个SparkSession对象，读取文件，失去一个DataSet弹性数据集对象。
ini复制代码SparkSession session = SparkSession.builder().master(“local[*]”).getOrCreate();
Dataset<Row> reader = session.read().text(“F:/test.txt”);
reader.show();

这里的show()就是打印输出以后数据集，它是一个action类的算子。
失去后果：
diff复制代码+———————–+

value
张三,20,研发部,普通员工
李四,31,研发部,普通员工
李丽,36,财务部,普通员工
张伟,38,研发部,经理
杜航,25,人事部,普通员工
周歌,28,研发部,普通员工

当初咱们拿到了根底数据，咱们应用map一对一操作，将一行行数据转换为Employee对象。
咱们这里不应用lamda表达式，让大家看得更加清晰。
这里实现了MapFunction接口里的call办法，每次拿到一行数据，咱们这里进行切分，再转换为对象。

须要特地指出的一点是，与后端WEB利用有一个对立异样解决不同的是，大数据利用，特地是流式计算，要保障7*24在线，须要对每个算子进行异样捕捉。
因为你不晓得上游数据荡涤到底怎么样，很可能拿到一条脏数据，解决的时候抛出异样，如果没有捕捉解决，那么整个利用就会挂掉。

spark的算子分为Transformation和Action两种类型。Transformation会开成一个DAG图，具备lazy提早性，它只会从一个dataset(rdd/df)转换成另一个dataset(rdd/df)，只有当遇到action类的算子才会真正执行。
咱们明天会演示的算子都是Transformation类的算子。
典型的Action算子包含show,collect,save之类的。比方在本地进行show查看后果，或者实现运行后save到数据库，或者HDFS。

spark执行时分为driver和executor。但不是本文的重点，不会开展讲。
只须要留神driver端会将代码散发到各个分布式系统的节点executor上，它自身不会参加计算。一般来说，算子内部，如以下示例代码的a处会在driver端执行，b处算子外部会不同服务器上的executor端执行。
所以在算子内部定义的变量，在算子外部应用的时候要特地留神！！不要想当然地认为都是一个main办法里写的代码，就肯定会在同一个JVM里。
这里波及到序列化的问题，同时它们分处不同的JVM，应用”==”比拟的时候也可能会出问题！！
这是一个后端WEB开发转向大数据开发时，这个思维肯定要转变过去。
简言之，后端WEB服务的分布式是咱们本人实现的，大数据的分布式是框架天生帮咱们实现的。

1.2.1 MapFunction
scss复制代码// a 算子内部，driver端
Dataset<Employee> employeeDataset = reader.map(new MapFunction<Row, Employee>() {

        @Override
        public Employee call(Row row) throws Exception {
            // b 算子外部，executor端
            Employee employee = null;
            try {
                // gson.fromJson(); 这里应用gson波及到序列化问题
                List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
            } catch (Exception exception) {
                // 日志记录
                // 流式计算中要做到7*24小时不间断，任意一条上流脏数据都可能导致失败，从而导致工作退出，所以这里要做好异样的抓取
                exception.printStackTrace();
            }
            return employee;
        }
    }, Encoders.bean(Employee.class));

    employeeDataset.show();

输入
diff复制代码+—+———-+——–+—-+

age	department	level	name
20	研发部	普通员工	张三
31	研发部	普通员工	李四
36	财务部	普通员工	李丽
38	研发部	经理	张伟
25	人事部	普通员工	杜航
28	研发部	普通员工	周歌

1.2.2 MapPartitionsFunction
spark中 map和mapPartitions有啥区别？
map是1条1条解决数据
mapPartitions是一个分区一个分区解决数据

后者肯定比前者效率高吗？
不肯定，看具体情况。
这里应用后面 map 一样的逻辑解决。能够看到在call办法里失去的是一个Iterator迭代器，是一批数据。
失去一批数据，而后再一对一映射为对象，再以Iterator的模式返回这批数据。
vbnet复制代码Dataset<Employee> employeeDataset2 = reader.mapPartitions(new MapPartitionsFunction<Row, Employee>() {

        @Override
        public Iterator<Employee> call(Iterator<Row> iterator) throws Exception {
            List<Employee> employeeList = new ArrayList<>();
            while (iterator.hasNext()){
                Row row = iterator.next();
                try {
                    List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                    Employee employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                    employeeList.add(employee);
                } catch (Exception exception) {
                    // 日志记录
                    // 流式计算中要做到7*24小时不间断，任意一条上流脏数据都可能导致失败，从而导致工作退出，所以这里要做好异样的抓取
                    exception.printStackTrace();
                }
            }
            return employeeList.iterator();
        }
    }, Encoders.bean(Employee.class));

    employeeDataset2.show();

输入后果跟map一样，这里就不贴出来了。
2.flatMap类
map和flatMap有什么区别？
map是一对一，flatMap是一对多。
当然在java stream中，flatMap叫法叫做扁平化。
这种思维，在java和spark,flink都是统一的。
2.1 java stream flatMap
以下代码将1条原始数据映射到2个对象上并返回。
ini复制代码List<Employee> employeeList2 = list.stream().flatMap(word -> {

        List<String> words = Arrays.stream(word.split(",")).collect(Collectors.toList());
        List<Employee> lists = new ArrayList<>();
        Employee employee = new Employee(words.get(0), Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        lists.add(employee);
        Employee employee2 = new Employee(words.get(0)+"_2", Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        lists.add(employee2);
        return lists.stream();
    }).collect(Collectors.toList());
    employeeList2.forEach(System.out::println);

输入
ini复制代码JavaStreamDemo.Employee(name=张三, age=20, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=张三_2, age=20, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=李四_2, age=31, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=李丽, age=36, department=财务部, level=普通员工)
JavaStreamDemo.Employee(name=李丽_2, age=36, department=财务部, level=普通员工)
JavaStreamDemo.Employee(name=张伟, age=38, department=研发部, level=经理)
JavaStreamDemo.Employee(name=张伟_2, age=38, department=研发部, level=经理)
JavaStreamDemo.Employee(name=杜航, age=25, department=人事部, level=普通员工)
JavaStreamDemo.Employee(name=杜航_2, age=25, department=人事部, level=普通员工)
JavaStreamDemo.Employee(name=周歌, age=28, department=研发部, level=普通员工)
JavaStreamDemo.Employee(name=周歌_2, age=28, department=研发部, level=普通员工)

2.2 spark flatMap
这里实现FlatMapFunction的call办法，一次拿到1条数据，而后返回值是Iterator，所以能够返回多条。
scss复制代码Dataset<Employee> employeeDatasetFlatmap = reader.flatMap(new FlatMapFunction<Row, Employee>() {

        @Override
        public Iterator<Employee> call(Row row) throws Exception {
            List<Employee> employeeList = new ArrayList<>();
            try {
                List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                Employee employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                employeeList.add(employee);

                Employee employee2 = new Employee(list.get(0)+"_2", Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                employeeList.add(employee2);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
            return employeeList.iterator();
        }
    }, Encoders.bean(Employee.class));
    employeeDatasetFlatmap.show();

输入
diff复制代码+—+———-+——–+——+

age	department	level	name
20	研发部	普通员工	张三
20	研发部	普通员工	张三_2
31	研发部	普通员工	李四
31	研发部	普通员工	李四_2
36	财务部	普通员工	李丽
36	财务部	普通员工	李丽_2
38	研发部	经理	张伟
38	研发部	经理	张伟_2
25	人事部	普通员工	杜航
25	人事部	普通员工	杜航_2
28	研发部	普通员工	周歌
28	研发部	普通员工	周歌_2

3 groupby类
与SQL相似，java stream流和spark一样，groupby对数据集进行分组并在此基础上能够进行聚合函数操作。也能够分组间接失去一组子数据集。
3.1 java stream groupBy
按部门分组统计部门人数：
ini复制代码Map<String, Long> map = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment, Collectors.counting()));

    System.out.println(map);

输入
复制代码{财务部=1, 人事部=1, 研发部=4}

3.2 spark groupBy
将映射为对象的数据集按部门分组，在此基础上统计部门员工数和平均年龄。
sql复制代码RelationalGroupedDataset datasetGroupBy = employeeDataset.groupBy(“department”);
// 统计每个部门有多少员工
datasetGroupBy.count().show();
/**

每个部门的平均年龄
*/

datasetGroupBy.avg(“age”).withColumnRenamed(“avg(age)”,”avgAge”).show();

输入别离为
diff复制代码+———-+—–+

department	count
财务部	1
人事部	1
研发部	4

diff复制代码+———-+——+

department	avgAge
财务部	36.0
人事部	25.0
研发部	29.25

3.3 spark groupByKey
spark的groupBy和groupByKey的区别,前者在此基础上应用聚合函数失去一个聚合值，后者只是进行分组，不进行任何计算。
相似于java stream的：
ini复制代码Map<String, List<Employee>> map2 = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment));
System.out.println(map2);

输入
css复制代码{财务部=[JavaStreamDemo.Employee(name=李丽, age=36, department=财务部, level=普通员工)],
人事部=[JavaStreamDemo.Employee(name=杜航, age=25, department=人事部, level=普通员工)],
研发部=[JavaStreamDemo.Employee(name=张三, age=20, department=研发部, level=普通员工), JavaStreamDemo.Employee(name=李四, age=31, department=研发部, level=普通员工), JavaStreamDemo.Employee(name=张伟, age=38, department=研发部, level=经理), JavaStreamDemo.Employee(name=周歌, age=28, department=研发部, level=普通员工)]}

应用spark groupByKey。
先失去一个key-value的一对多的一个汇合数据集。
这里的call()办法返回的是key,即分组的key。
java复制代码KeyValueGroupedDataset keyValueGroupedDataset = employeeDataset.groupByKey(new MapFunction<Employee, String>() {

        @Override
        public String call(Employee employee) throws Exception {
            // 返回分组的key,这里示意依据部门进行分组
            return employee.getDepartment();
        }
    }, Encoders.STRING());

再在keyValueGroupedDataset 的根底上进行mapGroups，在call()办法里就能够拿到每个key的所有原始数据。
vbnet复制代码keyValueGroupedDataset.mapGroups(new MapGroupsFunction() {

        @Override
        public Object call(Object key, Iterator iterator) throws Exception {
            System.out.println("key = " + key);
            while (iterator.hasNext()){
                System.out.println(iterator.next());
            }
            return iterator; 
        }
    }, Encoders.bean(Iterator.class))
            .show(); // 这里的show()没有意义，只是触发计算而已

输入
ini复制代码key = 人事部
SparkDemo.Employee(name=杜航, age=25, department=人事部, level=普通员工)
key = 研发部
SparkDemo.Employee(name=张三, age=20, department=研发部, level=普通员工)
SparkDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)
SparkDemo.Employee(name=张伟, age=38, department=研发部, level=经理)
SparkDemo.Employee(name=周歌, age=28, department=研发部, level=普通员工)
key = 财务部
SparkDemo.Employee(name=李丽, age=36, department=财务部, level=普通员工)

4 reduce类
reduce的字面意思是：缩小；减小；升高；放大。
又叫归约。
它将数据集进行循环，让以后对象和前一对象两两进行计算，每次计算失去的后果作为下一次计算的前一对象，并最终失去一个对象。
假如有5个数据【1，2，3，4，5】，应用reduce进行求和计算，别离是

比方下面的测试数据集，我要计算各部门年龄总数。应用聚合函数失去的是一个int类型的数字。
4.1 java stream reduce
scss复制代码int age = employeeList.stream().mapToInt(e -> e.age).sum();
System.out.println(age);//178

应用reduce也可进行下面的计算
scss复制代码int age1 = employeeList.stream().mapToInt(e -> e.getAge()).reduce(0,(a,b) -> a+b);
System.out.println(age1);// 178

然而我将年龄求和，同时失去一个残缺的对象呢？
ini复制代码JavaStreamDemo.Employee(name=周歌, age=178, department=研发部, level=普通员工)

能够应用reduce将数据集两两循环，将年龄相加，同时返回最初一个遍历的对象。
上面代码的pre 代表前一个对象,current 代表以后对象。
scss复制代码 /**

pre 代表前一个对象
current 代表以后对象
*/

Employee reduceEmployee = employeeList.stream().reduce(new Employee(), (pre,current) -> {

 // 当第一次循环时前一个对象为null
if (pre.getAge() == null) {
    current.setAge(current.getAge());
} else {
    current.setAge(pre.getAge() + current.getAge());
}
return current;

});
System.out.println(reduceEmployee);

4.2 spark reduce
spark reduce的根本思维跟java stream是一样的。
间接看代码：
ini复制代码 Employee datasetReduce = employeeDataset.reduce(new ReduceFunction<Employee>() {

        @Override
        public Employee call(Employee t1, Employee t2) throws Exception {
            // 不同的版本看是否须要判断t1 == null
            t2.setAge(t1.getAge() + t2.getAge());
            return t2;
        }
    });

    System.out.println(datasetReduce);

输入
ini复制代码SparkDemo.Employee(name=周歌, age=178, department=研发部, level=普通员工)

其它常见操作类
ini复制代码Employee employee = employeeDataset.filter(“age > 30”).limit(3).sort(“age”).first();
System.out.println(employee);
// SparkDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)

同时能够将dataset注册成table，应用更为弱小的SQL来进行各种弱小的运算。
当初SQL是flink的一等公民，spark也不遑多让。
这里举一个非常简单的例子。
css复制代码employeeDataset.registerTempTable(“table”);
session.sql(“select * from table where age > 30 order by age desc limit 3”).show();

输入
diff复制代码+—+———-+——–+—-+

age	department	level	name
38	研发部	经理	张伟
36	财务部	普通员工	李丽
31	研发部	普通员工	李四

scss复制代码employeeDataset.registerTempTable(“table”);
session.sql(“select

        concat_ws(',',collect_set(name)) as names, // group_concat
        avg(age) as age,
        department from table 
        where age > 30  
        group by department 
        order by age desc 
        limit 3").show();

输入
diff复制代码+———+—-+———-+

names	age	department
李丽	36.0	财务部
张伟,李四	34.5	研发部

小结
本文根据java stream的相似性，介绍了spark外面一些常见的算子操作。
本文只是做一个非常简单的入门介绍。
如果感兴趣的话，
后端的同学能够尝试着操作一下，非常简单，本地不须要搭建环境，只有引入spark 的 maven依赖即可。
我把本文的所有代码全副贴在最初面。
java stream 源码：

点击查看代码
ini复制代码import lombok.*;
import org.apache.commons.io.FileUtils;

import java.io.File;
import java.io.IOException;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

public class JavaStreamDemo {

public static void main(String[] args) throws IOException {
    /**
     * 张三,20,研发部,普通员工
     * 李四,31,研发部,普通员工
     * 李丽,36,财务部,普通员工
     * 张伟,38,研发部,经理
     * 杜航,25,人事部,普通员工
     * 周歌,28,研发部,普通员工
     */
    List<String> list = FileUtils.readLines(new File("f:/test.txt"), "utf-8");
    List<Employee> employeeList = list.stream().map(word -> {
        List<String> words = Arrays.stream(word.split(",")).collect(Collectors.toList());
        Employee employee = new Employee(words.get(0), Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        return employee;
    }).collect(Collectors.toList());

    // employeeList.forEach(System.out::println);

    List<Employee> employeeList2 = list.stream().flatMap(word -> {
        List<String> words = Arrays.stream(word.split(",")).collect(Collectors.toList());
        List<Employee> lists = new ArrayList<>();
        Employee employee = new Employee(words.get(0), Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        lists.add(employee);
        Employee employee2 = new Employee(words.get(0)+"_2", Integer.parseInt(words.get(1)), words.get(2), words.get(3));
        lists.add(employee2);
        return lists.stream();
    }).collect(Collectors.toList());
    // employeeList2.forEach(System.out::println);

    Map<String, Long> map = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment, Collectors.counting()));
    System.out.println(map);
    Map<String, List<Employee>> map2 = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment));
    System.out.println(map2);

    int age = employeeList.stream().mapToInt(e -> e.age).sum();
    System.out.println(age);// 178

    int age1 = employeeList.stream().mapToInt(e -> e.getAge()).reduce(0,(a,b) -> a+b);
    System.out.println(age1);// 178

    /**
     * pre 代表前一个对象
     * current 代表以后对象
     */
    Employee reduceEmployee = employeeList.stream().reduce(new Employee(), (pre,current) -> {
        if (pre.getAge() == null) {
            current.setAge(current.getAge());
        } else {
            current.setAge(pre.getAge() + current.getAge());
        }
        return current;
    });
    System.out.println(reduceEmployee);




}

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@ToString
static
class Employee implements Serializable {
    private String name;
    private Integer age;
    private String department;
    private String level;
}

}

spark的源码：

点击查看代码
java复制代码import com.google.gson.Gson;
import lombok.*;
import org.apache.spark.api.java.function.*;
import org.apache.spark.sql.*;

import java.io.Serializable;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Iterator;
import java.util.List;
import java.util.stream.Collectors;

public class SparkDemo {

public static void main(String[] args) {
    SparkSession session = SparkSession.builder().master("local[*]").getOrCreate();
    Dataset<Row> reader = session.read().text("F:/test.txt");
    // reader.show();
    /**
     * +-----------------------+
     * |                  value|
     * +-----------------------+
     * |张三,20,研发部,普通员工|
     * |李四,31,研发部,普通员工|
     * |李丽,36,财务部,普通员工|
     * |张伟,38,研发部,经理|
     * |杜航,25,人事部,普通员工|
     * |周歌,28,研发部,普通员工|
     * +-----------------------+
     */

    // 本地演示而已，理论分布式环境，这里的gson波及到序列化问题
    // 算子以外的代码都在driver端运行
    // 任何算子以内的代码都在executor端运行,即会在不同的服务器节点上执行
    Gson gson = new Gson();
    // a 算子内部，driver端
    Dataset<Employee> employeeDataset = reader.map(new MapFunction<Row, Employee>() {
        @Override
        public Employee call(Row row) throws Exception {
            // b 算子外部，executor端
            Employee employee = null;
            try {
                // gson.fromJson(); 这里应用gson波及到序列化问题
                List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
            } catch (Exception exception) {
                // 日志记录
                // 流式计算中要做到7*24小时不间断，任意一条上流脏数据都可能导致失败，从而导致工作退出，所以这里要做好异样的抓取
                exception.printStackTrace();
            }
            return employee;
        }
    }, Encoders.bean(Employee.class));

    // employeeDataset.show();
    /**
     * +---+----------+--------+----+
     * |age|department|   level|name|
     * +---+----------+--------+----+
     * | 20|    研发部|普通员工|张三|
     * | 31|    研发部|普通员工|李四|
     * | 36|    财务部|普通员工|李丽|
     * | 38|    研发部|    经理|张伟|
     * | 25|    人事部|普通员工|杜航|
     * | 28|    研发部|普通员工|周歌|
     */

    Dataset<Employee> employeeDataset2 = reader.mapPartitions(new MapPartitionsFunction<Row, Employee>() {
        @Override
        public Iterator<Employee> call(Iterator<Row> iterator) throws Exception {
            List<Employee> employeeList = new ArrayList<>();
            while (iterator.hasNext()){
                Row row = iterator.next();
                try {
                    List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                    Employee employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                    employeeList.add(employee);
                } catch (Exception exception) {
                    // 日志记录
                    // 流式计算中要做到7*24小时不间断，任意一条上流脏数据都可能导致失败，从而导致工作退出，所以这里要做好异样的抓取
                    exception.printStackTrace();
                }
            }
            return employeeList.iterator();
        }
    }, Encoders.bean(Employee.class));

    // employeeDataset2.show();
    /**
     * +---+----------+--------+----+
     * |age|department|   level|name|
     * +---+----------+--------+----+
     * | 20|    研发部|普通员工|张三|
     * | 31|    研发部|普通员工|李四|
     * | 36|    财务部|普通员工|李丽|
     * | 38|    研发部|    经理|张伟|
     * | 25|    人事部|普通员工|杜航|
     * | 28|    研发部|普通员工|周歌|
     * +---+----------+--------+----+
     */

    Dataset<Employee> employeeDatasetFlatmap = reader.flatMap(new FlatMapFunction<Row, Employee>() {
        @Override
        public Iterator<Employee> call(Row row) throws Exception {
            List<Employee> employeeList = new ArrayList<>();
            try {
                List<String> list = Arrays.stream(row.mkString().split(",")).collect(Collectors.toList());
                Employee employee = new Employee(list.get(0), Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                employeeList.add(employee);

                Employee employee2 = new Employee(list.get(0)+"_2", Integer.parseInt(list.get(1)), list.get(2), list.get(3));
                employeeList.add(employee2);
            } catch (Exception exception) {
                exception.printStackTrace();
            }
            return employeeList.iterator();
        }
    }, Encoders.bean(Employee.class));

// employeeDatasetFlatmap.show();

    /**
     * +---+----------+--------+------+
     * |age|department|   level|  name|
     * +---+----------+--------+------+
     * | 20|    研发部|普通员工|  张三|
     * | 20|    研发部|普通员工|张三_2|
     * | 31|    研发部|普通员工|  李四|
     * | 31|    研发部|普通员工|李四_2|
     * | 36|    财务部|普通员工|  李丽|
     * | 36|    财务部|普通员工|李丽_2|
     * | 38|    研发部|    经理|  张伟|
     * | 38|    研发部|    经理|张伟_2|
     * | 25|    人事部|普通员工|  杜航|
     * | 25|    人事部|普通员工|杜航_2|
     * | 28|    研发部|普通员工|  周歌|
     * | 28|    研发部|普通员工|周歌_2|
     * +---+----------+--------+------+
     */

    RelationalGroupedDataset datasetGroupBy = employeeDataset.groupBy("department");
    // 统计每个部门有多少员工
    // datasetGroupBy.count().show();
    /**
     * +----------+-----+
     * |department|count|
     * +----------+-----+
     * |    财务部|    1|
     * |    人事部|    1|
     * |    研发部|    4|
     * +----------+-----+
     */
    /**
     * 每个部门的平均年龄
     */
    // datasetGroupBy.avg("age").withColumnRenamed("avg(age)","avgAge").show();
    /**
     * +----------+--------+
     * |department|avg(age)|
     * +----------+--------+
     * |    财务部|    36.0|
     * |    人事部|    25.0|
     * |    研发部|   29.25|
     * +----------+--------+
     */

    KeyValueGroupedDataset keyValueGroupedDataset = employeeDataset.groupByKey(new MapFunction<Employee, String>() {
        @Override
        public String call(Employee employee) throws Exception {
            // 返回分组的key,这里示意依据部门进行分组
            return employee.getDepartment();
        }
    }, Encoders.STRING());

    keyValueGroupedDataset.mapGroups(new MapGroupsFunction() {
        @Override
        public Object call(Object key, Iterator iterator) throws Exception {
            System.out.println("key = " + key);
            while (iterator.hasNext()){
                System.out.println(iterator.next());
            }
            return iterator;
            /**
             * key = 人事部
             * SparkDemo.Employee(name=杜航, age=25, department=人事部, level=普通员工)
             * key = 研发部
             * SparkDemo.Employee(name=张三, age=20, department=研发部, level=普通员工)
             * SparkDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)
             * SparkDemo.Employee(name=张伟, age=38, department=研发部, level=经理)
             * SparkDemo.Employee(name=周歌, age=28, department=研发部, level=普通员工)
             * key = 财务部
             * SparkDemo.Employee(name=李丽, age=36, department=财务部, level=普通员工)
             */
        }
    }, Encoders.bean(Iterator.class))
            .show(); // 这里的show()没有意义，只是触发计算而已


    Employee datasetReduce = employeeDataset.reduce(new ReduceFunction<Employee>() {
        @Override
        public Employee call(Employee t1, Employee t2) throws Exception {
            // 不同的版本看是否须要判断t1 == null
            t2.setAge(t1.getAge() + t2.getAge());
            return t2;
        }
    });

    System.out.println(datasetReduce);


    Employee employee = employeeDataset.filter("age > 30").limit(3).sort("age").first();
    System.out.println(employee);
    // SparkDemo.Employee(name=李四, age=31, department=研发部, level=普通员工)

    employeeDataset.registerTempTable("table");
    session.sql("select * from table where age > 30 order by age desc limit 3").show();

    /**
     * +---+----------+--------+----+
     * |age|department|   level|name|
     * +---+----------+--------+----+
     * | 38|    研发部|    经理|张伟|
     * | 36|    财务部|普通员工|李丽|
     * | 31|    研发部|普通员工|李四|
     * +---+----------+--------+----+
     */


}

@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@ToString
public static class Employee implements Serializable {
    private String name;
    private Integer age;
    private String department;
    private String level;
}

}

spark maven依赖,自行不须要的spark-streaming,kafka依赖去掉。

关于springboot:揭开神秘面纱会stream流就会大数据

评论

发表回复取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

关于springboot:揭开神秘面纱会stream流就会大数据

评论

发表回复 取消回复

更多文章

苹果iOS打包的ipa应用无法安装？一篇文章带你了解可能的原因及排查方法

图解Golang：从零开始实现简易版过期LRU缓存

深入解析：基于Delta的线性数据结构模型，打造高效富文本编辑器

轻松管理社交媒体：使用Automa插件实现一键拉黑功能

发表回复取消回复