揭開神祕面紗，會 stream 流就會大數據
作者：是奉壹呀
原文：https://juejin.cn/post/7226612646543818807
如果你會任意一門語言的 stream 流，沒道理不會大數據開發。
俗話說男追女隔座山，女追男隔層紗。 如果說零基礎學大數據，感覺前面是一座山，那麼只要你會 java 或者任意一門語言的 stream 流，那大數據就只隔了一層紗。
準備工作
張三,20,研發部,普通員工
李四,31,研發部,普通員工
李麗,36,財務部,普通員工
張偉,38,研發部,經理
杜航,25,人事部,普通員工
周歌,28,研發部,普通員工@Getter
@Setter
@AllArgsConstructor
@NoArgsConstructor
@ToString
static
class Employee implements Serializable {
private String name;
private Integer age;
private String department;
private String level;
}因爲 scala 確實是比較小衆的語言，本文還是使用 java 演示 spark 代碼。
map 類
java stream map
map 表示一對一操作。將上游數據的一行數據進行任意操作，最終得到操作後的一條數據。 這種思想，在 java 和 spark，flink 都是一致的。
我們先用 java stream 演示讀取文件，再使用 map 操作將每行數據映射爲Employee對象。
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);轉換後的數據：
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=普通員工)spark map
首先得到一個 SparkSession 對象，讀取文件，得到一個 DataSet 彈性數據集對象。
SparkSession session = SparkSession.builder().master("local[*]").getOrCreate();
Dataset<Row> reader = session.read().text("F:/test.txt");
reader.show();這裏的 show() 就是打印輸出當前數據集，它是一個 action 類的算子。 得到結果：
+-----------------------+
| value|
+-----------------------+
|張三,20,研發部,普通員工|
|李四,31,研發部,普通員工|
|李麗,36,財務部,普通員工|
| 張偉,38,研發部,經理|
|杜航,25,人事部,普通員工|
|周歌,28,研發部,普通員工|
+-----------------------+這裏實現了 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服務的分佈式是我們自己實現的，大數據的分佈式是框架天生幫我們實現的。
MapFunction
// 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| 研發部|普通員工|周歌|MapPartitionsFunction
spark 中 map 和 mapPartitions 有啥區別？
map 是 1 條 1 條處理數據。 mapPartitions 是一個分區一個分區處理數據。
後者一定比前者效率高嗎？
不一定，看具體情況。
這裏使用前面 map 一樣的邏輯處理。可以看到在 call 方法裏得到的是一個 Iterator 迭代器，是一批數據。
得到一批數據，然後再一對一映射爲對象，再以 Iterator 的形式返回這批數據。
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 一樣，這裏就不貼出來了。
flatMap 類
map 和 flatMap 有什麼區別？
map 是一對一，flatMap 是一對多。 當然在 java stream 中，flatMap 叫法叫做扁平化。
這種思想，在 java 和 spark,flink 都是一致的。
java stream flatMap
以下代碼將 1 條原始數據映射到 2 個對象上並返回。
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);輸出
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=普通員工)spark flatMap
這裏實現 FlatMapFunction 的 call 方法，一次拿到 1 條數據，然後返回值是 Iterator，所以可以返回多條。
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|
+---+----------+--------+------+groupby 類
與 SQL 類似，java stream 流和 spark 一樣，groupby 對數據集進行分組並在此基礎上可以進行聚合函數操作。也可以分組直接得到一組子數據集。
java stream groupBy
按部門分組統計部門人數：
Map<String, Long> map = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment, Collectors.counting()));
System.out.println(map);輸出
{財務部=1, 人事部=1, 研發部=4}spark groupBy
將映射爲對象的數據集按部門分組，在此基礎上統計部門員工數和平均年齡。
RelationalGroupedDataset datasetGroupBy = employeeDataset.groupBy("department");
// 統計每個部門有多少員工
datasetGroupBy.count().show();
/**
* 每個部門的平均年齡
*/
datasetGroupBy.avg("age").withColumnRenamed("avg(age)","avgAge").show();輸出分別爲
+----------+-----+
|department|count|
+----------+-----+
| 財務部| 1|
| 人事部| 1|
| 研發部| 4|
+----------+-----+
+----------+------+
|department|avgAge|
+----------+------+
| 財務部| 36.0|
| 人事部| 25.0|
| 研發部| 29.25|
+----------+------+spark groupByKey
spark 的groupBy和groupByKey的區別，前者在此基礎上使用聚合函數得到一個聚合值，後者只是進行分組，不進行任何計算。
類似於 java stream 的：
Map<String, List<Employee>> map2 = employeeList.stream().collect(Collectors.groupingBy(Employee::getDepartment));
System.out.println(map2);輸出
{財務部=[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。
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 的所有原始數據。
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()沒有意義，只是觸發計算而已輸出
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=普通員工)reduce 類
reduce的字面意思是：減少；減小；降低；縮小。 又叫歸約。
它將數據集進行循環，讓當前對象和前一對象兩兩進行計算，每次計算得到的結果作爲下一次計算的前一對象，並最終得到一個對象。
假設有 5 個數據【1，2，3，4，5】，使用 reduce 進行求和計算，分別是
比如上面的測試數據集，我要計算各部門年齡總數。使用聚合函數得到的是一個 int 類型的數字。
java stream reduce
int age = employeeList.stream().mapToInt(e -> e.age).sum();
System.out.println(age);//178使用 reduce 也可進行上面的計算
int age1 = employeeList.stream().mapToInt(e -> e.getAge()).reduce(0,(a,b) -> a+b);
System.out.println(age1);// 178但是我將年齡求和，同時得到一個完整的對象呢？
JavaStreamDemo.Employee(name=周歌, age=178, department=研發部, level=普通員工)可以使用 reduce 將數據集兩兩循環，將年齡相加，同時返回最後一個遍歷的對象。
下面代碼的 pre 代表前一個對象，current 代表當前對象。
/**
* 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);spark reduce
spark reduce 的基本思想跟 java stream 是一樣的。
直接看代碼：
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);輸出
SparkDemo.Employee(name=周歌, age=178, department=研發部, level=普通員工)其它常見操作類
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 也不遑多讓。 這裏舉一個非常簡單的例子。
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| 研發部|普通員工|李四|
+---+----------+--------+----+
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();輸出
+---------+----+----------+
| names| age|department|
+---------+----+----------+
| 李麗|36.0| 財務部|
|張偉,李四|34.5| 研發部|
+---------+----+----------+小結
本文依據 java stream 的相似性，介紹了 spark 裏面一些常見的算子操作。
本文只是做一個非常簡單的入門介紹。
如果感興趣的話， 後端的同學可以嘗試着操作一下，非常簡單，本地不需要搭建環境，只要引入 spark 的 maven 依賴即可。
我把本文的所有代碼全部貼在最後面。
java stream 源碼：
點擊查看代碼
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 的源碼：
點擊查看代碼
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 依賴去掉。
點擊查看代碼
<properties>
<maven.compiler.source>8</maven.compiler.source>
<maven.compiler.target>8</maven.compiler.target>
<maven.compiler.source>1.8</maven.compiler.source>
<maven.compiler.target>1.8</maven.compiler.target>
<scala.version>2.12.15</scala.version>
<spark.version>3.2.0</spark.version>
<encoding>UTF-8</encoding>
</properties>
<dependencies>
<!-- scala依賴-->
<dependency>
<groupId>org.scala-lang</groupId>
<artifactId>scala-library</artifactId>
<version>${scala.version}</version>
</dependency>
<!-- spark依賴-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<version>1.18.2</version>
<scope>provided</scope>
</dependency>
<!--<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
<version>${spark.version}</version>
</dependency>-->
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-sql-kafka-0-10_2.12</artifactId>
<version>${spark.version}</version>
</dependency>
<dependency>
<groupId>com.google.code.gson</groupId>
<artifactId>gson</artifactId>
<version>2.7</version>
</dependency>
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.34</version>
</dependency>
</dependencies>本文由 Readfog 進行 AMP 轉碼，版權歸原作者所有。
來源：https://mp.weixin.qq.com/s/yuGUQF4v3W9Cg6k9gZYYBg