⼀⽂学完Flink流计算常⽤算⼦（Flink算⼦⼤全）
Flink和Spark类似，也是⼀种⼀站式处理的框架；既可以进⾏批处理（DataSet），也可以进⾏实时处理（DataStream）。
所以下⾯将Flink的算⼦分为两⼤类：⼀类是DataSet，⼀类是DataStream。
DataSet
⼀、Source算⼦
1. fromCollection
fromCollection：从本地集合读取数据
例：
val env = ExecutionEnvironment
val textDataSet: DataSet[String] = env.fromCollection(
List("1，张三", "2，李四", "3，王五", "4，赵六")
)
2. readTextFile
readTextFile：从⽂件中读取：
val textDataSet: DataSet[String] = adTextFile("/")
3. readTextFile：遍历⽬录
readTextFile可以对⼀个⽂件⽬录内的所有⽂件，包括所有⼦⽬录中的所有⽂件的遍历访问⽅式：
gasstationval parameters = new Configuration
// umeration 开启递归
parameters.tBoolean("umeration", true)
val file = adTextFile("/data").withParameters(parameters)
4. readTextFile：读取压缩⽂件
对于以下压缩类型，不需要指定任何额外的inputformat⽅法，flink可以⾃动识别并且解压。但是，压缩⽂件可能不会并⾏读取，可能是顺序读取的，这样可能会影响作业的可伸缩性。
压缩⽅法⽂件扩展名是否可并⾏读取
DEFLATE.deflate no
< .gzip no
Bzip2.bz2no
< no
val file = adTextFile("/")
⼆、Transform转换算⼦
numb是什么意思
因为Transform算⼦基于Source算⼦操作，所以⾸先构建Flink执⾏环境及Source算⼦，后续Transform算⼦操作基于此：
val env = ExecutionEnvironment
和平村英语
val textDataSet: DataSet[String] = env.fromCollection(
List("张三,1", "李四,2", "王五,3", "张三,4")
)
1. map
将DataSet中的每⼀个元素转换为另外⼀个元素：
// 使⽤map将List转换为⼀个Scala的样例类
ca class Ur(name: String, id: String)
val urDataSet: DataSet[Ur] = textDataSet.map {
text =>
val fieldArr = text.split(",")
Ur(fieldArr(0), fieldArr(1))
}
urDataSet.print()
2. flatMap
将DataSet中的每⼀个元素转换为0...n个元素：
// 使⽤flatMap操作，将集合中的数据：
// 根据第⼀个元素，进⾏分组
// 根据第⼆个元素，进⾏聚合求值
val result = textDataSet.flatMap(line => line)
.groupBy(0) // 根据第⼀个元素，进⾏分组
.sum(1) // 根据第⼆个元素，进⾏聚合求值
result.print()
3. mapPartition
将⼀个分区中的元素转换为另⼀个元素：
// 使⽤mapPartition操作，将List转换为⼀个scala的样例类
ca class Ur(name: String, id: String)
val result: DataSet[Ur] = textDataSet.mapPartition(line => {
line.map(index => Ur(index._1, index._2))
})
result.print()
4. filter
过滤出来⼀些符合条件的元素，返回boolean值为true的元素：
val source: DataSet[String] = env.fromElements("java", "scala", "java")
val filter:DataSet[String] = source.filter(line => ains("java"))//过滤出带java的数据filter.print()
5. reduce
可以对⼀个datat或者⼀个group来进⾏聚合计算，最终聚合成⼀个元素：
// 使⽤ fromElements 构建数据源
val source = env.fromElements(("java", 1), ("scala", 1), ("java", 1))
// 使⽤map转换成DataSet元组
val mapData: DataSet[(String, Int)] = source.map(line => line)
// 根据⾸个元素分组
val groupData = upBy(_._1)皮肤怎么补水
// 使⽤reduce聚合
val reduceData = duce((x, y) => (x._1, x._2 + y._2))
/
/ 打印测试
reduceData.print()
6. reduceGroup
将⼀个datat或者⼀个group聚合成⼀个或多个元素。
reduceGroup是reduce的⼀种优化⽅案；
它会先分组reduce，然后在做整体的reduce；这样做的好处就是可以减少⽹络IO：
// 使⽤ fromElements 构建数据源
val source: DataSet[(String, Int)] = env.fromElements(("java", 1), ("scala", 1), ("java", 1))
// 根据⾸个元素分组
val groupData = upBy(_._1)
// 使⽤reduceGroup聚合
val result: DataSet[(String, Int)] = duceGroup {
(in: Iterator[(String, Int)], out: Collector[(String, Int)]) =>
val tuple = in.reduce((x, y) => (x._1, x._2 + y._2))
}
// 打印测试
result.print()
7. minBy和maxBy
选择具有最⼩值或最⼤值的元素：greenlee
// 使⽤minBy操作，求List中每个⼈的最⼩值
// List("张三,1", "李四,2", "王五,3", "张三,4")
ca class Ur(name: String, id: String)
// 将List转换为⼀个scala的样例类
val text: DataSet[Ur] = textDataSet.mapPartition(line => {
line.map(index => Ur(index._1, index._2))
})
val result = text
.groupBy(0) // 按照姓名分组
.minBy(1) // 每个⼈的最⼩值
8. Aggregate
在数据集上进⾏聚合求最值（最⼤值、最⼩值）：
val data = new mutable.MutableList[(Int, String, Double)]
data.+=((1, "yuwen", 89.0))
data.+=((2, "shuxue", 92.2))
data.+=((3, "yuwen", 89.99))
// 使⽤ fromElements 构建数据源
val input: DataSet[(Int, String, Double)] = env.fromCollection(data)
// 使⽤group执⾏分组操作
val value = upBy(1)
// 使⽤aggregate求最⼤值元素
.aggregate(Aggregations.MAX, 2)
// 打印测试
value.print()
Aggregate只能作⽤于元组上
注意：
要使⽤aggregate，只能使⽤字段索引名或索引名称来进⾏分组 groupBy(0) ，否则会报⼀下错误:
Exception in thread "main" java.lang.UnsupportedOperationException: Aggregate does not support grouping with KeySelector functions, yet.
9. distinct
去除重复的数据：
// 数据源使⽤上⼀题的
// 使⽤distinct操作，根据科⽬去除集合中重复的元组数据
val value: DataSet[(Int, String, Double)] = input.distinct(1)
value.print()
10. first
取前N个数：
input.first(2) // 取前两个数
11. join
将两个DataSet按照⼀定条件连接到⼀起，形成新的DataSet：
// s1 和 s2 数据集格式如下：
// DataSet[(Int, String,String, Double)]
摄氏度英文val joinData = s1.join(s2) // s1数据集 join s2数据集
.where(0).equalTo(0) { // join的条件
(s1, s2) => (s1._1, s1._2, s2._2, s1._3)
}
12. leftOuterJoin
左外连接，左边的Datat中的每⼀个元素，去连接右边的元素
此外还有：
npirightOuterJoin：右外连接，左边的Datat中的每⼀个元素，去连接左边的元素
fullOuterJoin：全外连接，左右两边的元素，全部连接
下⾯以 leftOuterJoin 进⾏⽰例：
on the wayval data1 = ListBuffer[Tuple2[Int,String]]()
data1.append((1,"zhangsan"))
data1.append((2,"lisi"))
data1.append((3,"wangwu"))
data1.append((4,"zhaoliu"))
val data2 = ListBuffer[Tuple2[Int,String]]()
guangzhoudata2.append((1,"beijing"))
data2.append((2,"shanghai"))
data2.append((4,"guangzhou"))
val text1 = env.fromCollection(data1)
val text2 = env.fromCollection(data2)
text1.leftOuterJoin(text2).where(0).equalTo(0).apply((first,cond)=>{
if(cond==null){
(first._1,first._2,"null")
}el{
(first._1,first._2,cond._2)
}
宾语补足语
}).print()
13. cross
交叉操作，通过形成这个数据集和其他数据集的笛卡尔积，创建⼀个新的数据集
和join类似，但是这种交叉操作会产⽣笛卡尔积，在数据⽐较⼤的时候，是⾮常消耗内存的操作：
val cross = ss(input2){
(input1 , input2) => (input1._1,input1._2,input1._3,input2._2)
}
cross.print()
14. union
联合操作，创建包含来⾃该数据集和其他数据集的元素的新数据集，不会去重：
val unionData: DataSet[String] = elements1.union(elements2).union(elements3)
/
/ 去除重复数据
val value = unionData.distinct(line => line)
15. rebalance
Flink也有数据倾斜的时候，⽐如当前有数据量⼤概10亿条数据需要处理，在处理过程中可能会发⽣如图所⽰的状况：
