[TOC]
1.wordcount
利用socket作为数据源，对输入的每行数据进行单词计数。计算频率为process time的每10秒一次，结果输出到terminal。
1object SocketWindowWordCount { 2 def main(args: Array[String]) : Unit = { 3 4 val port: Int = try { 5 ParameterTool.fromArgs(args).getInt("port") 6 } catch { 7 case e: Exception => { 8 System.err.println("No port specified. Please run 'xx.jar --port <port>'") 9 return 10 } 11 } 12 13 val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment 14 15 val text = env.socketTextStream("localhost", port, '\n') 16 17 val windowCounts = text 18 .flatMap(_.split("\\s")) 19 .map(WordWithCount(_,1)) 20 .keyBy(_.word) 21 .window(TumblingProcessingTimeWindows.of(Time.seconds(10))) 22 .sum("count") 23 24 windowCounts.print() 25 26 env.execute("Socket Window WordCount") 27 } 28 29 case class WordWithCount(word: String, count: Long) 30}
数据格式
1case class SensorReading(id: String, timestamp: Long, temperature: Double) 2 3object SmokeLevel extends Enumeration { 4 type SmokeLevel = SmokeLevel.Value 5 val High, Low = Value 6} 7 8case class Alert(message: String, timestamp: Long)
2.双流警报EventTime
时间特征为event time，每1s更新一次watermark，watermark由SensorReading内部的timestamp推进，允许5s的延迟（过滤掉迟到数据）。数据源SensorReading并发处理，数据源SmokeLevel并发度为1，但能够被每个并发的SensorReading流访问。假设两个流的数据是源源不断的。当SensorReading的temperature大于100且SmokeLevel为High时触发警报。警报包含当时SensorReading的timestamp。
下面例子，迟到数据，即数据晚于WM依然会被处理
注意：如果某个流在connect前assignTimestampsAndWatermarks，connect后的流是不会更新WM的。
1def main(args: Array[String]): Unit = { 2 3 val env = StreamExecutionEnvironment.getExecutionEnvironment 4 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 5 env.getConfig.setAutoWatermarkInterval(1000L) 6 7 val sensorInput = env 8 .addSource(new SensorSource) 9 .assignTimestampsAndWatermarks( 10 new BoundedOutOfOrdernessTimestampExtractor[SensorReading](Time.seconds(5)) { 11 override def extractTimestamp(element: SensorReading): Long = { 12 element.timestamp 13 } 14 }) 15 val smokeLevelInput = env 16 .addSource(new SmokeLevelSource) 17 .setParallelism(1) 18 19 val res = sensorInput 20 .process(new MyKeyedProcessFunction) // 这里的实现省略，其实就是if和collect 21 .connect(smokeLevelInput) 22 .flatMap(new MyFlatMapFunc) 23 24 res.print() 25 26 env.execute("multiple streamss") 27 28} 29 30class MyFlatMapFunc extends CoFlatMapFunction[SensorReading, SmokeLevel, Alert] { 31 private private var curSmokeLevel = SmokeLevel.Low 32 33 override def flatMap1(value: SensorReading, out: Collector[Alert]): Unit = { 34 if (curSmokeLevel.equals(SmokeLevel.High) && value.temperature > 100) { 35 out.collect(Alert("Alert! ", value.timestamp)) 36 } 37 } 38 39 override def flatMap2(value: SmokeLevel, out: Collector[Alert]): Unit = { 40 curSmokeLevel = value 41 } 42}
3.持续计数stateful + timer + SideOutputs
对每个key的数据量进行累加计数，如果1分钟没有新数据，就输出key-count对。对每一个数据进行处理时，sideoutput当前所处理的key的state数据（更新后的）
1val realTimeInfo: OutputTag[String] = 2 new OutputTag[String]("real-time_info") 3 4def main(args: Array[String]): Unit = { 5 val env = StreamExecutionEnvironment.getExecutionEnvironment 6 env.getConfig.setAutoWatermarkInterval(1000L) 7 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 8 9 val countStream = env.addSource(new SensorSource) 10 .keyBy(_.id) 11 .process(new MyProcessFunc) 12 13 countStream.getSideOutput(realTimeInfo) 14 .print() 15 16 env.execute() 17} 18 19case class CountWithTimeStamp(key: String, count: Int, ts: Long) 20 21class MyProcessFunc extends KeyedProcessFunction[String, SensorReading, (String, Int)] { 22 23 lazy val state = getRuntimeContext 24 .getState(new ValueStateDescriptor[CountWithTimeStamp]("muState", classOf[CountWithTimeStamp])) 25 26 override def processElement(value: SensorReading, 27 ctx: KeyedProcessFunction[String, SensorReading, (String, Int)]#Context, 28 out: Collector[(String, Int)]): Unit = { 29 val current = state.value() match { 30 case null => 31 CountWithTimeStamp(value.id, 1, ctx.timestamp()) 32 case CountWithTimeStamp(key, count, lastModified) => 33 // 删除上一次的timer 34 ctx.timerService().deleteEventTimeTimer(lastModified + 6000) 35 CountWithTimeStamp(key, count + 1, ctx.timestamp()) 36 } 37 38 state.update(current) 39 40 ctx.timerService().registerEventTimeTimer(current.ts + 6000) 41 42 ctx.output(realTimeInfo, current) 43 } 44 45 override def onTimer(timestamp: Long, 46 ctx: KeyedProcessFunction[String, SensorReading, (String, Int)]#OnTimerContext, 47 out: Collector[(String, Int)]): Unit = { 48 state.value() match { 49 case CountWithTimeStamp(key, count, lastModified) => 50 if (timestamp == lastModified) out.collect((key, count)) else None 51 case _ => 52 } 53 } 54}
4.一定时间范围内的极值windowfunction + checkpoint
利用tumbling window计算各个sensor在15s内的最大最小值，返回结果包含窗口的结束时间。另外，window只存储极值，不保留原数据。
checkpoint间隔为10s，watermark刷新间隔1s
1def main(args: Array[String]): Unit = { 2 val env = StreamExecutionEnvironment.getExecutionEnvironment 3 env.getConfig.setAutoWatermarkInterval(1000) 4 env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime) 5 6 env.enableCheckpointing(10 * 1000L) 7 8 env.addSource(new SensorSource) 9 .assignTimestampsAndWatermarks(new MyPeriodicAssigner) 10 .map(r => (r.id, r.temperature, r.temperature)) 11 .keyBy(_._1) 12 .window(TumblingEventTimeWindows.of(Time.seconds(15))) 13 .reduce( 14 (item1: (String, Double, Double), item2: (String, Double, Double)) => { 15 (item1._1, item1._2.min(item2._2), item1._3.max(item2._3)) 16 }, 17 new MyWindowEndProcessFunction() 18 ) 19} 20 21case class MaxMinTemperature(key: String, min: Double, max: Double, ts: Long) 22 23class MyWindowEndProcessFunction 24 extends ProcessWindowFunction[(String, Double, Double), MaxMinTemperature, String, TimeWindow] { 25 override def process(key: String, context: Context, elements: Iterable[(String, Double, Double)], 26 out: Collector[MaxMinTemperature]): Unit = { 27 out.collect(MaxMinTemperature(key, elements.head._2, elements.head._3, context.window.getEnd)) 28 } 29}