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 = input．groupBy(1)
// 使用aggregate求最大值元素
．aggregate(Aggregations．MAX, 2)
// 打印测试
value．print()
Aggregate只能作用于元组上
注意:
要使用aggregate,只能使用字段索引名或索引名称来进行分组 groupBy(0) ,否则会报一下错误:
Exception in thread "main" java．lang．UnsupportedOperationException: Aggregate does not supportgrouping 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
左外连接,左边的Dataset中的每一个元素,去连接右边的元素
此外还有:
rightOuterJoin:右外连接,左边的Dataset中的每一个元素,去连接左边的元素
fullOuterJoin:全外连接,左右两边的元素,全部连接
下面以 leftOuterJoin 进行示例:
val 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]]()
data2．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,second)=>{
if(second==null){
(first．_1,first．_2,"null")
}else{
(first．_1,first．_2,second．_2)
}
})．print()
13． cross
交叉操作,通过形成这个数据集和其他数据集的笛卡尔积,创建一个新的数据集
和join类似,但是这种交叉操作会产生笛卡尔积,在数据比较大的时候,是非常消耗内存的操作
val cross = input1．cross(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亿条数据需要处理,在处理过程中可能会发生如图所示的状况:
这个时候本来总体数据量只需要10分钟解决的问题,出现了数据倾斜,机器1上的任务需要4个小时才能完成,那么其他3台机器执行完毕也要等待机器1执行完毕后才算整体将任务完成;所以在实际的工作中,出现这种情况比较好的解决方案就是接下来要介绍的—rebalance(内部使用round robin方法将数据均匀打散。这对于数据倾斜时是很好的选择。)
// 使用rebalance操作,避免数据倾斜
val rebalance = filterData．rebalance()
16． partitionByHash
按照指定的key进行hash分区
val data = new mutable．MutableList[(Int, Long, String)]
data．+=((1, 1L, "Hi"))
data．+=((2, 2L, "Hello"))
data．+=((3, 2L, "Hello world"))
val collection = env．fromCollection(data)
val unique = collection．partitionByHash(1)．mapPartition{
line =>
line．map(x => (x．_1 , x．_2 , x．_3))
}
unique．writeAsText("hashPartition", WriteMode．NO_OVERWRITE)
env．execute()
17． partitionByRange
根据指定的key对数据集进行范围分区
val data = new mutable．MutableList[(Int, Long, String)]
data．+=((1, 1L, "Hi"))
data．+=((2, 2L, "Hello"))
data．+=((3, 2L, "Hello world"))
data．+=((4, 3L, "Hello world, how are you?"))
val collection = env．fromCollection(data)
val unique = collection．partitionByRange(x => x．_1)．mapPartition(line => line．map{
x=>
(x．_1 , x．_2 , x．_3)
})
unique．writeAsText("rangePartition", WriteMode．OVERWRITE)
env．execute()
18． sortPartition
根据指定的字段值进行分区的排序
val data = new mutable．MutableList[(Int, Long, String)]
data．+=((1, 1L, "Hi"))
data．+=((2, 2L, "Hello"))
data．+=((3, 2L, "Hello world"))
data．+=((4, 3L, "Hello world, how are you?"))
val ds = env．fromCollection(data)
val result = ds
．map { x => x }．setParallelism(2)
．sortPartition(1, Order．DESCENDING)//第一个参数代表按照哪个字段进行分区
．mapPartition(line => line)
．collect()
println(result)
三、Sink算子1． collect
将数据输出到本地集合
result．collect()
2． writeAsText
将数据输出到文件
Flink支持多种存储设备上的文件,包括本地文件,hdfs文件等
Flink支持多种文件的存储格式,包括text文件,CSV文件等
// 将数据写入本地文件
result．writeAsText("/data/a", WriteMode．OVERWRITE)
// 将数据写入HDFS
result．writeAsText("hdfs://node01:9000/data/a", WriteMode．OVERWRITE)
DataStream流处理算子
和DataSet一样,DataStream也包括一系列的Transformation操作
一、Source算子
Flink可以使用 StreamExecutionEnvironment．addSource(source) 来为我们的程序添加数据来源。
Flink 已经提供了若干实现好了的 source functions,当然我们也可以通过实现 SourceFunction 来自定义非并行的source或者实现 ParallelSourceFunction 接口或者扩展 RichParallelSourceFunction 来自定义并行的 source。
Flink在流处理上的source和在批处理上的source基本一致。大致有4大类:
基于本地集合的source(Collection-based-source)基于文件的source(File-based-source)- 读取文本文件,即符合 TextInputFormat 规范的文件,并将其作为字符串返回基于网络套接字的source(Socket-based-source)- 从 socket 读取。元素可以用分隔符切分。自定义的source(Custom-source)烹饪1-375
下面使用addSource将Kafka数据写入Flink为例:
如果需要外部数据源对接,可使用addSource,如将Kafka数据写入Flink,先引入依赖:
将Kafka数据写入Flink:
val properties = new Properties()
properties．setProperty("bootstrap．servers", "localhost:9092")
properties．setProperty("group．id", "consumer-group")
properties．setProperty("key．deserializer", "org．apache．kafka．common．serialization．StringDeserializer")
properties．setProperty("value．deserializer", "org．apache．kafka．common．serialization．StringDeserializer")
properties．setProperty("auto．offset．reset", "latest")
val source = env．addSource(new FlinkKafkaConsumer011[String]("sensor", new SimpleStringSchema(), properties))
基于网络套接字的:
val source = env．socketTextStream("IP", PORT)
二、Transform转换算子1． map
将DataSet中的每一个元素转换为另外一个元素
dataStream．map { x => x * 2 }
2． FlatMap
采用一个数据元并生成零个,一个或多个数据元。将句子分割为单词的flatmap函数
dataStream．flatMap { str => str．split(" ") }
3． Filter
计算每个数据元的布尔函数,并保存函数返回true的数据元。过滤掉零值的过滤器
dataStream．filter { _ != 0 }
4． KeyBy
逻辑上将流分区为不相交的分区。具有相同Keys的所有记录都分配给同一分区。在内部,keyBy()是使用散列分区实现的。指定键有不同的方法。
此转换返回KeyedStream,其中包括使用被Keys化状态所需的KeyedStream。
dataStream．keyBy(0)
5． Reduce
被Keys化数据流上的“滚动”Reduce。将当前数据元与最后一个Reduce的值组合并发出新值
keyedStream．reduce { _ + _ }
6． Fold
具有初始值的被Keys化数据流上的“滚动”折叠。将当前数据元与最后折叠的值组合并发出新值
val result: DataStream[String] = keyedStream．fold("start")((str, i) => { str + "-" + i })
// 解释:当上述代码应用于序列(1,2,3,4,5)时,输出结果“start-1”,“start-1-2”,“start-1-2-3”,．．．
7． Aggregations
在被Keys化数据流上滚动聚合。min和minBy之间的差异是min返回最小值,而minBy返回该字段中具有最小值的数据元(max和maxBy相同)。
keyedStream．sum(0);
keyedStream．min(0);
keyedStream．max(0);
keyedStream．minBy(0);
keyedStream．maxBy(0);
8． Window
可以在已经分区的KeyedStream上定义Windows。Windows根据某些特征(例如,在最后5秒内到达的数据)对每个Keys中的数据进行分组。这里不再对窗口进行详解,有关窗口的完整说明,请查看这篇文章:Flink 中极其重要的 Time 与 Window 详细解析
dataStream．keyBy(0)．window(TumblingEventTimeWindows．of(Time．seconds(5)));
9． WindowAll
Windows可以在常规DataStream上定义。Windows根据某些特征(例如,在最后5秒内到达的数据)对所有流事件进行分组。
注意:在许多情况下,这是非并行转换。所有记录将收集在windowAll 算子的一个任务中。
dataStream．windowAll(TumblingEventTimeWindows．of(Time．seconds(5)))
10． Window Apply
将一般函数应用于整个窗口。
注意:如果您正在使用windowAll转换,则需要使用AllWindowFunction。
下面是一个手动求和窗口数据元的函数
windowedStream．apply { WindowFunction }
allWindowedStream．apply { AllWindowFunction }
11． Window Reduce
将函数缩减函数应用于窗口并返回缩小的值
windowedStream．reduce { _ + _ }
12． Window Fold
将函数折叠函数应用于窗口并返回折叠值
val result: DataStream[String] = windowedStream．fold("start", (str, i) => { str + "-" + i })
// 上述代码应用于序列(1,2,3,4,5)时,将序列折叠为字符串“start-1-2-3-4-5”
13． Union
两个或多个数据流的联合,创建包含来自所有流的所有数据元的新流。注意:如果将数据流与自身联合,则会在结果流中获取两次数据元
dataStream．union(otherStream1, otherStream2, ．．．)
14． Window Join
在给定Keys和公共窗口上连接两个数据流
dataStream．join(otherStream)
．where(
在给定的时间间隔内使用公共Keys关联两个被Key化的数据流的两个数据元e1和e2,以便e1．timestamp + lowerBound <= e2．timestamp <= e1．timestamp + upperBound
am．intervalJoin(otherKeyedStream)
．between(Time．milliseconds(-2), Time．milliseconds(2))
．upperBoundExclusive(true)
．lowerBoundExclusive(true)
．process(new IntervalJoinFunction() {．．．})
16． Window CoGroup
在给定Keys和公共窗口上对两个数据流进行Cogroup
dataStream．coGroup(otherStream)
．where(0)．equalTo(1)
．window(TumblingEventTimeWindows．of(Time．seconds(3)))
．apply (new CoGroupFunction () {．．．})
17． Connect
“连接”两个保存其类型的数据流。连接允许两个流之间的共享状态
DataStream
类似于连接数据流上的map和flatMap
connectedStreams．map(
(_ : Int) => true,
(_ : String) => false)connectedStreams．flatMap(
(_ : Int) => true,
(_ : String) => false)
19． Split
根据某些标准将流拆分为两个或更多个流
val split = someDataStream．split(
(num: Int) =>
(num % 2) match {
case 0 => List("even")
case 1 => List("odd")
})
20． Select
从拆分流中选择一个或多个流
