开发者学堂课程【开源 Flink 极客训练营：Stream Processing with Apache Flink】学习笔记，与课程紧密联系，让用户快速学习知识。
课程地址：https://developer.aliyun.com/learning/course/760/detail/13338
Stream Processing with Apache Flink
1、时间相关 API
目标 Event Time Processing Time
获取记录事件 context.getTimestamp()或从数据字段中获取 timerService().currentProcessingTime()
获取“watermark” timerService().currentWatermark()
timerService().currentProcessingTime()
注册定时器 timerService.registerEventTimeTimer()
timerService.registerProcessingTimeTimer
在编写逻辑的过程中所用到的时间相关的 API ，还是按照 Event Time 还有 Processing Time 进行区分。在应用逻辑里通过接口对应的支持，可以做三件事，第一件事是获取记录的时间来，可以调context.getTimestamp 或者如果是 SQL、Table 会直接从数据字段中把对应的时间提取出来， Processing Time 是直接掉currentProcessingTime 就可以，service 的内部是直接调用从获取系统时间的静态方法来返回值，除了获取记录的时间，还需要获取watermark ，把 watermark 加引号，因为只有Event Time 里才有 watermark 的概念，而 Processing Time 里没有 watermark概念，如果非要把一个东西当成 watermark，就是本身数据时间，比如第一次调用 timerservice.current Processing Time 方法之后获取一个值，既是当前记录的时间也是当前的 watermark 值，因为时间往前流，第一次调用值之后，第二次调用值不会比第一次值还小。第三个是可以在一些相关的方法里去注册一些定时器，作用是可以比如有一个 cache，对于 cache 在今后的某一个时间进行类似于清理或者其他的工作，清理应该发生在将来的某一个时间点，可以调用timerservice.registerEvent Time Timer 或者 Processing Time Timer 方法注册定时器之后，要在整个的方法里去添加对于定时器回调的处理逻辑，每当对应的流程 Event Time 或者 Processing Time 的时间超过定时器时间之后，就会调用方法里去自己编写的定时器回调逻辑，可以把对于 cache 的清理或者一些其他的工作放到逻辑里面去完成。
例子：
/**
* 7. Processing time tumbling window.
*/
public static void presessingTinchindew() throws Exception {
StreamExecutionEnviromment e=StreamExecutionEnvirormen t.getExecutionEnvironment();
DataStreamSourcesInteger> source =e
.add5ourcelnew 5ourceFunction-Integer>(){
private volatile boolean stop = false;
@Override
public void runlSourceContext ctx]throws Exception {
int1=8;
white (1stop 5& 1 < data.size()) {
ctx.collect(data.get(i++));
Thread.sleep(200);
}
}
@Override
public void cancell) { stop =true; }
}).setParallelism(1);
e.setStreanTineCharacteristic(TineCharacteristic.ProcessingTime);source.keyBy(wv2).processnewKeyedProcessFunction(){
private static final int wINDOwSIZE= 200;
private TreeMapcLong，Integerwindows;
@Override
public void open(Configaration parameters)throws Exception{
super.open(paraneters)；
windows=new TreeMap-o()；
@Override
public void processElement(Integer valve, Context ctx, Collector out] {
Long currentTime =ctx.timerServicel).currentProcessingTime(); long windowstart =currentTime / WINDOWSIZE;
// Update the window
int sum= windows.getOrDefault(windowStart,0)
windows.put(windowstart,sum+value)；
// Fire old windows
Mapoldwindows=windows.headMap(windowStart;false);
Iterator
out.collect(iterator.next().getValue());
iterator.remove()；
}
}
@Override
public void close() throws Exception {
super.close();
Systen.out.println(windows);
}
}）.print().setParallelism(2);
e.execute();
}
自己定义 tubling window 翻滚的的窗口，窗口和窗口之间没有重合，是基于 Processing Time 完成，第一步是获取运行环境，然后新添加数据的源，没有直接用刚才定的 data 集合里去取数据，因为涉及到时间，所以为了看出来效果，需要在数据注入的过程当中人为的去引入一些延迟，基本的逻辑是遍历 data 的集合里边获取数据，每一次获取数据之后，都引入了200毫秒的延迟，这是简单的 Source Function，如果要用 Processing Time 要注意在环境里设置时间特性， process 是默认的特性。 Processing Time 是一行可有可无，但是为了表示效果把它加上，如果要用 Event Time 一定要把时间特性设置，设置成为 Event Time ，设置完了之后要从 Source Function 出发来进行。首先进行按照奇偶数字进行分组，同样定义一个 keyed process function，具体在 Keyed process function 第一步是为了把它和状态区分开，这里没有用系统内部 Flink 里状态的方法，直接简单使用 treemap 存储所有的温度，温度的 size 设成200毫米，在 open 方法里首先对于 treemap 用来存取不同窗口里时间。存取不同的窗口里的数值的数据结构进行一个初始化，初始化之后每到一条数据，首先通过调用ctx.timerService.currentProcessingTime 获取当前的时间，获取时间之后，要对数据进行窗口的划分，划分用窗口的开始时间作为窗口唯一的标识进行划分，直接把当前的时间去除以窗口的大小，取证，可以得到窗口的开始时间。要对数据窗口里的数据进行 sum 操作，获取窗口现有的 sum 值之后，可以 treemap.getOrDefault 的方法，如果窗口里的值已经存在，就会返回已有的 sum 值，如果没有存在就回去。返回之后需要更新窗口里的数值，可以把新计算的结果退回去，回去之后还要进行一些判断，没有用到 timer 直接去在处理的逻辑里判断，把所有的小于窗口、旧的窗口拿出来去遍历旧的窗口，因为旧窗口里有的值肯定不会再更新，遍历一下值然后把结果发送出去，发送出去之后旧的窗口没有用了，直接删除。最后 close方法在整个处理完成之后会自动调用方法，在里边为了防止没有来得及触发的窗口的数值看不到，就把所有剩下的窗口里面剩下的值都输出来，最后把结果用 print 方法打印到控制台上把并行度设置为2。
执行一下结果为：
/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/bin/Java...
2>1
1>2
2>3
1>4
2>5
1>6
2>7
1>8
2>9
{}
{}
{}
{7969896895=10}
Process finished with exit code 0
窗口的数据源每发送一条数据都加上200毫秒的间隔，窗口大小也是200毫秒，所以基本上每过来一条数据会触发一个新的窗口。
把200毫秒改成400毫秒，演示结果：
/Library/Java/JavaVirtualMachines/jdk1.8.0_101.jdk/Contents/Home/bin/Java...
2>3
1>7
2>11
1>15
{3984948598=19}
{}
{}
{}
Process finished with exit code 0
