目录
自定义Source
Tranform
map
flatMap
Fliter
KeyBy
滚动聚合算子（Rolling Aggregation）
Reduce
Split 和 Select
Connect 和 CoMap
Union
Connect 与 Union 区别
自定义Source
flink自定义source接收数据
除了以上的 source 数据来源，我们还可以自定义 source。需要做的，只是传入 一个 SourceFunction 就可以。
val stream4 = env.addSource( new MySensorSource()具体调用如下：
//自定义SourceFuntion
class MySensorSource() extends SourceFunction[SensorReading]{//定义一个标识位flag，用来表示数据源是否正常运行发出数据var running: Boolean =trueoverride def cancel(): Unit = running =falseoverride def run(ctx: SourceFunction.SourceContext[SensorReading]): Unit = {//定义一个随机数发生器val rand = new Random()//随机生成一组（10个）传感器的初始温度:（id ，temp）var curTemp = 1.to(10).map(i => ("sensor_" + i, rand.nextDouble() * 100))//定义无线循环，不停地产生数据，除非被cancelwhile(running){//在上次数据基础上微调，更新温度值curTemp = curTemp.map(data => (data._1,data._2 + rand.nextGaussian()))//获取当前时间戳，加入到数据中val curTime = System.currentTimeMillis()curTemp.foreach(data => ctx.collect(SensorReading(data._1,curTime,data._2)))//间隔500msThread.sleep(500)}}
运行结果：
它会一直更新接收数据
Tranform
转换算子
map
val streamMap = stream.map { x => x*2 }flatMap
flatMap 的函数签名：def flatMap[A,B](as: List[A])(f: A ⇒ List[B]): List[B]
例如: flatMap(List(1,2,3))(i ⇒ List(i,i))
结果是 List(1,1,2,2,3,3),
而 List("a b", "c d").flatMap(line ⇒ line.split(" "))
结果是 List(a, b, c, d)。
val streamFlatMap = stream.flatMap{
x => x.split(" ")
}Fliter
val streamFilter = stream.filter{
x => x == 1
}
KeyBy
DataStream → KeyedStream：逻辑地将一个流拆分成不相交的分区，每个分 区包含具有相同 key 的元素，在内部以 hash
滚动聚合算子（Rolling Aggregation）
sum()
min()
max()
minBy()
maxBy()
Reduce
KeyedStream → DataStream：一个分组数据流的聚合操作，合并当前的元素 和上次聚合的结果，产生一个新的值，返回的流中包含每一次聚合的结果，而不是 只返回最后一次聚合的最终结果。
val stream2 = env.readTextFile("YOUR_PATH\\sensor.txt")
.map( data => {
val dataArray = data.split(",")
SensorReading(dataArray(0).trim, dataArray(1).trim.toLong,
dataArray(2).trim.toDouble)
})
.keyBy("id")
.reduce( (x, y) => SensorReading(x.id, x.timestamp + 1, y.temperature)Split 和 Select
Split
DataStream → SplitStream：根据某些特征把一个 DataStream 拆分成两个或者 多个 DataStream
Select
SplitStream→DataStream：从一个 SplitStream 中获取一个或者多个 DataStream。
需求：传感器数据按照温度高低（以 30 度为界），拆分成两个流
val splitStream = stream2.split( sensorData => {if (sensorData.temperature > 30) Seq("high") else Seq("low")
} )val high = splitStream.select("high")val low = splitStream.select("low")val all = splitStream.select("high", "low")Connect 和 CoMap
Connect
DataStream,DataStream → ConnectedStreams：连接两个保持他们类型的数 据流，两个数据流被 Connect 之后，只是被放在了一个同一个流中，内部依然保持 各自的数据和形式不发生任何变化，两个流相互独立。
CoMap
ConnectedStreams → DataStream：作用于 ConnectedStreams 上，功能与 map 和 flatMap 一样，对 ConnectedStreams 中的每一个 Stream 分别进行 map 和 flatMap 处理。
val warning = high.map( sensorData => (sensorData.id,
sensorData.temperature) )
val connected = warning.connect(low)val coMap = connected.map(
warningData => (warningData._1, warningData._2, "warning"),
lowData => (lowData.id, "healthy")
)Union
DataStream → DataStream：对两个或者两个以上的 DataStream 进行 union 操 作，产生一个包含所有 DataStream 元素的新 DataStream。
//合并以后打印
val unionStream: DataStream[StartUpLog] = appStoreStream.union(otherStream)
unionStream.print("union:::")Connect 与 Union 区别
1． Union 之前两个流的类型必须是一样，Connect 可以不一样，在之后的 coMap 中再去调整成为一样的。
2. Connect 只能操作两个流，Union 可以操作多个
