04
Event-time 真正的事件产生的时间。 事件日志记录受网络传输延迟，并不得到真正的事件产生时间。
时间三兄弟
- EventTime 事件真正产生的时间。这个字段是要带在数据里面的，不然无法获取。
- IngestionTime 进入引擎的时间，受网络影响，时间不确定。大小介于二者间。
- ProcessingTime 算子执行时的当前系统的时间，时间不确定
设置时间语义，按时间三兄弟中哪个来处理。
从1.12起，默认的时间语义从ProcessingTime改成了EventTime，如果数据源没有可记录EventTime的字段，或没有设置waterMark，就会报错。需要手动设置为ProcessingTime。
env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime)
Window
对流处理，数据是源源不断进入的，无界的（批处理是有界的）。不可能无界限一直一直计算，Window 把一个无界流数据进行切分，得到有界流（流转批），在窗口内进行一组计算。 （和 ss 的微批是相反的思想（批转流））
窗口的分类
根据是否有keyBy分两类
- Keyed Window ? 使用 stream.keyBy.window 划分窗口
- Non-Keyed Window ? 使用 stream.windowAll 划分窗口
Assigner
负责将每条数据分发到正确的window中去
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/datastream/operators/windows/#window-assigners
计数的 CountWindow
是根据元素的个数来划分，达到指定数量就划分。与时间是没关系。 如果keyby，要某个key达到指定数量才会统计这个key，而不统计其他key，只能其他key也达到指定数量才触发自己key的统计。
计时的 TimeWindow
达到时间长度就切分。
滚动窗口 TumblingWindow
window size可以是CountSize也可以是TimeSize（又既可以是EventTime也可以是ProcessingTime ）。不带key时用windowAll，带key用window。
滑动窗口 SlidingWindow
window size 和 slide size是不等的，window size是一个窗口的大小，代表这个窗口从开始道结束的间隔；slide size可以理解是前后两个窗口开始时间的间隔。一个元素可能会被落在多个window中；window间可能有重叠。可以用来分析一种趋势、走势。
如图示，假设每个window的大小是10s（window size），而每隔5s就开启下一个window（slide size）。则一个元素会被落在两个window中。
滚动窗口可以理解为window size 和 slide size相等的的滑动窗口，是滑动窗口的特例。上一个窗口结束同时开启才下一个窗口，所以不会重叠。
时间间隔 SessionWindow
只有没有操作的时间间隔达到指定size的间隔后，才开始算新window。数据不断，就一直是旧window（保持在当前window）
总结
WindowFunction
window之后，数据在窗口内进行每组window进行计算，如sum(0)。
增量：ReduceFunction, AggregateFunction 数据来了就会进行算的；但是不适合每项业务的，如排序、最值。
全量：ProcessWindowFunction 窗口OK了才进行计算 全量数据。
自定义 AggregateFunction，重写 createAccumulator()、add()、getResult()、merge()方法
05
自定义 ProcessFunction，重写 processElement()
Watermark 水印
理想情况下先操作的数据应该先到达，受网络原因，先产生的数据不一定先到，需要解决，以EventTime来确定数据顺序。
延迟(乱序)数据处理：不能等待数据全部到达再处理，等待时间不可预测，等待过久就变成了批处理。
乱序数据处理：window + Watermark 。Watermark 是衡量EventTime进展的机制。是一条特殊的数据记录。
使用：在获取环境时，设置为EventTime；获取数据源时同时指定watermark的字段（一般情况都在数据源时设置，但支持在transformation时）。
EventTimeSessionWindows
// 旧API
val lines = environment
.socketTextStream("gargantua", 9527)
// Time.seconds(0) 代表容忍度为0
.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor[String](Time.seconds(0)) {
// 设置哪个字段作为watermark的时间
override def extractTimestamp(element: String): Long = {
element.split(",")(0).toLong
}
})
// 新API
val lines = environment
.socketTextStream("gargantua", 9527)
.assignTimestampsAndWatermarks(WatermarkStrategy.forBoundedOutOfOrderness[String](Duration.ofSeconds(0))
.withTimestampAssigner(new SerializableTimestampAssigner[String] {
// 设置哪个字段作为watermark的时间
override def extractTimestamp(element: String, recordTimestamp: Long): Long = element.split(",")(0).toLong
}))
// 在transformation
.window(EventTimeSessionWindows.withGap(Time.seconds(5)))
之前学习SessionWindow时是以ProcessingTime做时间语义，就会以终端的时间来作为window划分的time size。如果以EventTime（watermark指定数据里某个字段作为EventTime），那么就会关联数据里的这个字段的时间作为划分window的依据（具体计算过程？）。
TumblingEventTimeWindows
需要注意，当前数据所携带的时间，>= 上一个window的结束边界的时间，则触发上一个window。但当前数据不会纳入上一个window的统计；它是属于下一个window的，作为下一个window的第一条数据。
延迟的数据（属于上一个window的范围，但上一个window已经被触发过了的数据）：
- 通过设置容忍度，如Time.seconds(2)代表可以允许数据延迟2s以内。即增加了2s的触发间隔，但不改变window的大小。例如window size为5s，则在>=第7s的数据才能触发一个window，但这个window统计数据的范围还是前5s的。只是允许了5-7s这个范围时属于1-5s的数据延迟仍然可以进入1-5s的window。超出容忍范围的数据仍然会被丢弃。
- 被丢弃的数据只能通过侧流输出。
// 侧流 side output (OutputTag的类型应该和侧输出流中的类型保持一致，否则会报错 type mismatch) ctx.output(new OutputTag[(String,Long,Float)]("high"),(value.name,value.time,value.temperature)) // 或 window.getSideOutput(outputTag).print("侧流输出...")
SlidingEventTimeWindows
.window(SlidingEventTimeWindows.of(Time.seconds(6), Time.seconds(2))) // 6s window size ; 2s slide size
滑动窗口每个window都是独立的触发，独立的统计自己范围的数据，只是可能会重复。
Flink on Yarn
Flink结构
flink client(提交flink作业) --> JobManager(主) --> TaskManager(从)
Standalone模式用于学习，生产都是 on Yarn 、on k8s。
Flink on Yarn 是不需要集群的，同Spark on Yarn 一样，Spark、Flink只作为客户端提交作业。
