自定义source
只需要传入一个SourceFunction即可
val stream4 = env.addSource( new MySensorSource() )
举例说明：随机生成传感器数据
无非就是通过生成随机数据的方式组装成传感器数据而已
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
DataStream、KeyedStream 逻辑地将一个流拆分成不相交的分区 每个分区包含相同key的元素 在内部以hash的形式实现
代码： val aggStream = dataStream.keyBy("id")
- 聚合滚动算子(Rolling Aggregation)
针对keyedStream的每个支流做聚合 sum() min() max() minBy() maxBy()
