Untitled
Posted on
1 | import org.apache.flink.streaming.api.scala._ |
Untitled
Posted on
How to Avoid the ReceiverDisconnectedException
1 | 21/08/19 02:18:45 ERROR TaskSetManager [task-result-getter-0]: Task 32 in stage 73.0 failed 4 times; aborting job |
flink window
Posted on
In
flink
flink1.6
Keyed Windows
1 | stream |
Non-Keyed Windows
1 | stream |
窗口生命周期
window
trigger
The function will contain the computation to be applied to the contents of the window, while the Trigger specifies the conditions under which the window is considered ready for the function to be applied. A triggering policy might be something like “when the number of elements in the window is more than 4”, or “when the watermark passes the end of the window”. A trigger can also decide to purge a window’s contents any time between its creation and removal. Purging in this case only refers to the elements in the window, and not the window metadata. This means that new data can still be added to that window.
function包含了窗口中数据的计算逻辑，trigger指定了在哪些条件下会调用这些function
Evictor
Apart from the above, you can specify an Evictor (see Evictors) which will be able to remove elements from the window after the trigger fires and before and/or after the function is applied.
flink数据源
Posted on
In
flink
flink窗口类
Posted on
In
flink
- WindowAssigner
- SlidingProcessingTimeWindows
- BaseAlignedWindowAssigner
- SlidingAlignedProcessingTimeWindows
- TumblingEventTimeWindows
- TumblingTimeWindows
- MergingWindowAssigner
- ProcessingTimeSessionWindows
- DynamicProcessingTimeSessionWindows
- DynamicEventTimeSessionWindows
- EventTimeSessionWindows
- TumblingProcessingTimeWindows
- SlidingEventTimeWindows
- SlidingTimeWindows
- GlobalWindows
WindowAssigner
1. SlidingProcessingTimeWindows
1 | /** |
6. SlidingEventTimeWindows
1 | /** |
6.1 SlidingTimeWindows(废弃)
2. BaseAlignedWindowAssigner
1.1 SlidingAlignedProcessingTimeWindows
5. TumblingProcessingTimeWindows
3. TumblingEventTimeWindows
1 | /** |
3.1 TumblingTimeWindows(废弃)
1 | /** |
4. MergingWindowAssigner
4.1 ProcessingTimeSessionWindows
4.2 DynamicProcessingTimeSessionWindows
4.3 DynamicEventTimeSessionWindows
4.4 EventTimeSessionWindows
7. GlobalWindows
总结
适用场景
- 滑动窗口
- 每条数据会发送到多个滑动窗口中，即在最终的输出中，一条数据要被统计多次
- 适合统计据当前时间往前一段时间内的数据汇总
- 滚动窗口
- 每条数据只会在一个滚动窗口中
- 适合对数据进行简单聚合后，再次聚合的场景
- 适合输出明细，不做聚合的场景，比如join后直接输出
占用内存比较
- 滑动窗口
- 理论上一条数据会复制到多个窗口，被复制几次，占用内存就会扩大几倍，但不清楚是否有优化，比如只复制数据的引用？
- 滚动窗口
关于时间点
窗口的开始时间
- 首先要明确的是，开始时间只与system time和offset参数相关，与程序开始运行时间无关
- 比如设置了窗口size是1h，那么在
窗口的结束时间
flink配置log4j
Posted on
因为本地测试中，本地构建的数据源并非预想中的单线程运行，所以想看下线程id，以确定是否每次都新起了线程。
推荐使用xml，结构清晰
1 | <?xml version="1.0" encoding="UTF-8"?> |
解析
如果Logger不指定additivity=false，默认为true。此时当前Logger及Root都会处理日志，如果Logger与Root配了相同的Appender，就会重复输出。
Root的level，表示的是Root使用的日志级别，如果有一个Logger没有具体指定，就会交由Root处理，即日志级别由Root决定。
flink官网目录
Posted on
In
flink
参考资料
Posted on
In
flink
flink实践问题汇总
Posted on
In
flink
待整理
- 本地自定义数据源，会启动多个线程，且不会停止，无法控制单线程输入
本地自定义数据源，运行起来不停止
日志记录
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:53 - loop:2
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:54 - operationInfo:{"operationTimeStamp":"2021-01-27 16:58:23","pageId":"30001","sessionId":"abcd","unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:47 - loop:2
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:48 - orderInfo:{"orderId":10001,"orderTimeStamp":"2021-01-27 16:58:23","price":39.8,"spuList":["spu1","spu2","spu3"],"unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:53 - loop:2
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:54 - operationInfo:{"operationTimeStamp":"2021-01-27 16:58:23","pageId":"30001","sessionId":"abcd","unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:47 - loop:2
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:48 - orderInfo:{"orderId":10001,"orderTimeStamp":"2021-01-27 16:58:23","price":39.8,"spuList":["spu1","spu2","spu3"],"unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:47 - loop:2
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:48 - orderInfo:{"orderId":10001,"orderTimeStamp":"2021-01-27 16:58:23","price":39.8,"spuList":["spu1","spu2","spu3"],"unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:53 - loop:2
2021-01-27 16:58:23 Source: operation_data_stream (1/1) ERROR LocalOperationSource:54 - operationInfo:{"operationTimeStamp":"2021-01-27 16:58:23","pageId":"30001","sessionId":"abcd","unionId":"123","userId":"user_123"}
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:47 - loop:2
2021-01-27 16:58:23 Source: order_data_stream (1/1) ERROR LocalOrderSource:48 - orderInfo:{"orderId":10001,"orderTimeStamp":"2021-01-27 16:58:23","price":39.8,"spuList":["spu1","spu2","spu3"],"unionId":"123","userId":"user_123"}
现象：
- 两个数据源，都不断输出loop:2的记录，不会执行到loop:1，如果删除掉
Thread.sleep()可以执行到loop:1 - 不会停止，按理说只应该执行一次loop:2
原因：
- 估计是
Thread.sleep()把线程挂起后，无法唤醒，所以执行不到loop:1 - 可以判断与数据源无关，是受后续的处理方式影响的，如果直接sink，不会有问题2