自定义source
只需要传入一个SourceFunction即可
val stream4 = env.addSource( new MySensorSource() )
举例说明：随机生成传感器数据
无非就是通过生成随机数据的方式组装成传感器数据而已
Transform
- 转换算子
val streamMap = stream.map { x => x * 2 }- flatMap
a、
flatMap(List(1,2,3))(i -> List(i,i)) 结果是 List(1,1,2,2,3,3)
b、
List("a b","c d").flatMap(line -> line.split(" ")) 结果是List(a,b,c,c)
代码：
val streamFlatMap = stream.flatMap{
x => x.split(" ")
}
- Filter
val streamFilter = stream.filter{
x => x == 1
}- KeyBy
DataStream、KeyedStream 逻辑地将一个流拆分成不相交的分区
每个分区包含相同key的元素 在内部以hash的形式实现
代码：
val aggStream = dataStream.keyBy("id")
- 聚合滚动算子(Rolling Aggregation)
针对keyedStream的每个支流做聚合
sum()
min()
max()
minBy()
maxBy()
- Reduce
源码
https://gitee.com/pingfanrenbiji/Flink-UserBehaviorAnalysis/blob/master/FlinkTutorial/src/main/scala/com/xdl/apitest/TransformTest.scala
KeyedStream、DataStream 一个分组数据流的聚合操作
合并当前的元素和上次聚合的结果 产生一个新的值
返回的流中包含每一次聚合的结果
而不是只返回最后一次聚合的最终结果
stream split
stream select
- ConnectedStreams
连接两个保持它们类型的数据流 两个数据流被Connect之后 只是被放在了同一个流中 内部依然保持各自的数据和形式不发生任何变化 两个流相互独立
- CoMap,CoFlatMap
作用于ConnectedStream上功能与map和flatMap一样
对ConnectedStreams中的每一个stream分别进行map和flatMap处理
- Union
对两个或两个以上的DataStream进行union操作 产生一个包含所有DataStram元素的新的DataStream
connect和union区别
1、union之前两个流的类型必须一样 connect可以不一样 在之后的coMap中再调整成一样的
2、Connect只能操作两个流 Union可以操作多个
支持的数据类型
Flink流应用程序处理的是以数据对象表示事件流
数据对象需要被序列化和反序列化
以便通过网络传送它们或从状态后端、检查点和保存点读取它们
Flink使用类型信息的概念来表示数据类型
并为每个数据类型生成特定的序列化器、反序列化器和比较器
Flink具有类型提取系统 该系统分析函数的输入和返回类型
以自动获取类型信息 从而获取序列化和反序列化器
lamdba 函数或泛型类型 需要显示的提供类型信息 才能使得应用程序正常工作或提高性能
基础数据类型
Flink支持Java和Scala中所常见的数据类型
- Long
val numbers: DataStream[Long] = env.fromElements(1L, 2L, 3L, 4L)
numbers.map( n => n + 1 )
- 元组
val persons: DataStream[(String, Integer)] = env.fromElements(
("Adam", 17),
("Sarah", 23) )
persons.filter(p => p._2 > 18)
- 样例类
case class Person(name: String, age: Int)
val persons: DataStream[Person] = env.fromElements(
Person("Adam", 17),
Person("Sarah", 23) )
persons.filter(p => p.age > 18)
- 简单类
public class Person {
public String name;
public int age;
public Person() {}
public Person(String name, int age) {
this.name = name;
this.age = age;
}
}
DataStream<Person> persons = env.fromElements(
new Person("Alex", 42),
new Person("Wendy", 23));- Flink 对 Java 和 Scala(ArrayList，HashMap，Enum等)也支持
实现UDF函数-更细粒度的控制流
函数类(Function classes)
Flink暴露了所有udf函数的接口(实现方式为接口或抽象类)
例如：
MapFunction、FilterFunction、ProcessFunction
- 实现FilterFunction接口
class FilterFilter extends FilterFunction[String] {
override def filter(value: String): Boolean = {
value.contains("flink")
}
}
val flinkTweets = tweets.filter(new FlinkFilter)- 将函数实现成匿名类
val flinkTweets = tweets.filter(
new RichFilterFunction[String] {
override def filter(value: String): Boolean = {
value.contains("flink")
}
}
)
- 字符串作为参数传进去
val tweets: DataStream[String] = ...
val flinkTweets = tweets.filter(new KeywordFilter("flink"))
class KeywordFilter(keyWord: String) extends FilterFunction[String] {
override def filter(value: String): Boolean = {
value.contains(keyWord)
}
}
匿名函数(Lamdba Functions)
val tweets: DataStream[String] = ...
val flinkTweets = tweets.filter(_.contains("flink"))
// _ .id 代表 data => data.id
dataStream.filter( _.id.statWith("sensor_1")).print
副函数
函数类接口 所有的Flink函数类都有Rich版本
与常规函数区别：
可以获取运行时上下文
并拥有一些生命周期方法
可以实现更加复杂的功能
a、 RichMapFunction
b、 RichFlatMapFunction
c、 RichFilterFunction
生命周期：
a、open() 是 rich fuction初始化方法
当一个算子例如map或filter被调用之前 open会被调用
b、close方法 是生命周期中的最后一个调用方法 做一些清理工作
c、getRuntimeContext方法 提供了函数的RuntimeContext的一些信息
例如函数的执行并行度、任务的名字以及state状态
- 代码
class MyFlatMap extends RichFlatMapFunction[Int, (Int, Int)] {
var subTaskIndex = 0
override def open(configuration: Configuration): Unit = {
subTaskIndex = getRuntimeContext.getIndexOfThisSubtask
// 以下可以做一些初始化工作 ， 例如建立一个和 HDFS 的连接
}
override def flatMap(in: Int, out: Collector[(Int, Int)]): Unit = {
if (in % 2 == subTaskIndex) {
out.collect((subTaskIndex, in))
}
}
override def close(): Unit = {
// 以下做一些清理工作，例如断开和 HDFS 的连接。
}
}
