1、张静/技术专家张芒/技术专家FLINK SQL 在快手的扩展和实践SQL在快手功能扩展性能优化稳定性提升未来展望#1#2#3#4#5#1FLINK SQL 在快手FLINK SQL 在快手30%全量占比30%增量占比4亿/s峰值吞吐-1240%60%6亿/s#2功能扩展功能扩展#1Group Window Aggregate 扩展1.支持多维分析2.引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWWindow Table-Valued Function 扩展(FLIP-145)1.丰富 Window TVF 算子2.支持 W

2、indow Offset3.支持 Batch Mode#2支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOW多维分析Group Window Aggregate 扩展1.标准的 GROUPING SETS/ROLLUP/CUBE 语法2.支持各种 GROUP WINDOW FUNCTION案例统计主题维度和总维度下的 UVSELECTtheme_id,COUNT(DISTINCT user_id)as uv,CUMULATE_END(ts,INTERVAL 1 MINUTE,INTERVAL 1 DAY)FROM source_tabl

3、eGROUP BYCUMULATE(ts,INTERVAL 1 MINUTE,INTERVAL 1 DAY),GROUPING SETS(),(theme_id)支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展业务背景绘制数据曲线，每个点表示当天0点到当前的累积指标历史方案1:无限流聚合，把时间归一到分钟粒度后，作为 Group Key 之一历史方案1的缺点：不满足输出到屏幕上的点不再变化的要求历史方案2:窗口聚合：Tumble Window（1天）+Ea

4、rly Fire。历史方案2的缺点：1.曲线上的每个点，时间和累积值在时间上没有对应关系 发生 failover 或者回溯历史，曲线不能还原 各个时间点上分维度的累计值加和不等于总维度的累计值2.在采用 Split Distinct Aggregation 避免数据倾斜时，自增曲线有凹坑支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展历史方案2 问题1：历史回溯，曲线不能还原且不光滑问题2：自增曲线有凹坑支持多维分析引入高阶窗口函数 CUMULATE WI

5、NDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展历史方案2 自增曲线有凹坑的原因：第一级聚合的结果是更新流支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展语法CUMULATE(time_attr,interval,interval,interval)time_attr:必选参数，时间属性列step interval:必选参数，定义窗口的步长size interval:必

6、选参数，定义一个窗口的 sizeoffset interval:可选参数，指定窗口开始的偏移量支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展语法CUMULATE(time_attr,interval,interval,interval)支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展案例绘制数据曲线，每分钟打一个点，

7、每个点表示当天0点到当前分钟的累积 UVSELECTCUMULATE_END(ts,INTERVAL 1 MINUTE,INTERVAL 1 DAY)AS window_time,page,COUNT(DISTINCT user_id)AS uvFROM source_tableGROUP BYCUMULATE(ts,INTERVAL 1 MINUTE,INTERVAL 1 DAY),page支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展案例绘制数据曲线

8、，每分钟打一个点，每个点表示当天0点到当前分钟的累积 UVCumulate Window 方案下，回溯历史的数据曲线支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWCUMULATE WINDOWGroup Window Aggregate 扩展Cumulate Window 方案的优点1.时间点取窗口的结束时间，数据严格按照事件时间划分窗口 每条输出数据里，时间点和累积值在时间上是严格对应的 回溯场景下，数据可还原且曲线平滑 各个时间点上分维度的累计值加和等于总维度的累计值2.输出 Append 流而不是 Update 流 在采用 S

9、plit Distinct Aggregation 避免数据倾斜时，自增曲线不会有凹坑支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWDYNAMIC CUMULATE WINDOWGroup Window Aggregate 扩展业务背景绘制数据曲线，每个点表示每个直播间自开播到当前的累积指标语法DYNAMIC_CUMULATE(time_attr,interval,interval)time_attr:必选参数，时间属性列step interval:必选参数，定义窗口的步长idle interval:可选参数，定义一个窗口的 gap

10、，指定当前 Key 多久没来数据就认为该窗口结束了支持多维分析引入高阶窗口函数 CUMULATE WINDOW DYNAMIC CUMULATE WINDOWDYNAMIC CUMULATE WINDOWGroup Window Aggregate 扩展案例绘制数据曲线，每个点表示每个直播间自开播以来到对应时间点的累积 UV，一个直播间如果连续 1小时没有数据流入，可以认为该直播间已关闭SELECTDYNAMIC_CUMULATE_END(ts,INTERVAL 1 MINUTE,INTERVAL 60 MINUTE),room_id,COUNT(DISTINCT user_id)as uvF

11、ROM source_tableGROUP BYroom_id,DYNAMIC_CUMULATE(ts,INTERVAL 1 MINUTE,INTERVAL 60 MINUTE)丰富 Window TVF 算子Window Table-valued Function 扩展丰富 Window TVF 算子支持 Window Offset支持 Batch ModeWindow TVF 算子集合1.Window Aggregation*2.Window TopN(1.13)3.Window Join(1.14)4.Window Deduplication(1.15)5.Individual Wind

12、ow TVF query(1.15)丰富 Window TVF 算子丰富 Window TVF 算子支持 Window Offset支持 Batch Mode案例统计0点到当前时间最热销商品中前100名的销售额和买家数以及爆品所属的主播下所有商品的买家数（案例引自 FLIP-145）Window Table-valued Function 扩展Window offset(1.14)丰富 Window TVF 算子支持 Window Offset支持 Batch Mode参数用途用于调整窗口的划分逻辑参数说明 可选参数，默认值是 0。Duration 类型，可以是正数，也可以是负数 不影响 wa

13、termark 对相同的窗口，不同的 offset 值可能产生相同的偏移效果。比如 SIZE 是 10 分钟的 Tumble 窗口，offset 设置为 6 MINUTE 和-4 MINUTE 的效果是一样的。Window Table-valued Function 扩展+6 min-4 minWindow offset(1.14)丰富 Window TVF 算子支持 Window Offset支持 Batch Mode案例统计每个页面从本周一 0点截止到当前时间点的累积 uv SELECTwindow_end,page,COUNT(DISTINCT user_id)AS uvFROM TAB

14、LE(CUMULATE(TABLE source_table,DESCRIPTOR(ts),INTERVAL 1 MINUTE,INTERVAL 7 DAY,INTERVAL 4 DAY)GROUP BY window_start,window_end,pageWindow Table-valued Function 扩展支持 Batch Mode(1.15)丰富 Window TVF 算子支持 Window Offset支持 Batch Mode目标所有在流模式可以运行的 Window TVF query，也可以用批模式运行。原理引入 Batch WTF 算子，给输入数据加上所属的窗口属性后

15、发给下游。下游算子复用已有的 Batch 模式的算子（HashAgg/SortAgg，HashJoin/SortMergeJoin），不需要 state，更加高效。Window Table-valued Function 扩展#3性能优化性能优化#1Group Window Aggregate Distinct MapState 状态复用维表 JOIN MiniBatch 维表 JOIN，减少 RPC 调用次数#2Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinDistinct State 状态复用性能

16、优化案例统计 APP 每个子频道的UV，该用例有两个业务特点：1.频道是可枚举的2.不同频道的用户重叠度很高SELECTchannel,COUNT(DISTINCT device_id)AS uvFROM source_tableGROUP BY channelGroup Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinDistinct State 状态复用性能优化每个频道有对应的 Distinct MapState，但不同频道的 MapState 有大量重合，如何复用这个 State？Group KeyMapSt

17、ate:KeyMapState:ValueA1121B1131OTHER31优化前的状态存储格式(MapState 是 Long 类型，64 bit)Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinDistinct State 状态复用性能优化历史方案基于社区提供的优化方法（Use FILTER Modifier on Distinct Aggregates）手动改写SQLSELECTTMP.channel,TMP.uvFROM(SELECTCOUNT(DISTINCT device_id)FILTER

18、(WHERE channel=A)AS uv_1,COUNT(DISTINCT device_id)FILTER(WHERE channel=B)AS uv_2,COUNT(DISTINCT device_id)FILTER(WHERE channel=OTHER)AS uv_3FROM source_table)T1,LATERAL TABLE(toMultipleRow(uv_1,uv_2,uv_3)AS TMP(channel,uv)Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinDistinct

19、 State 状态复用性能优化Group KeyMapState:KeyMapState:Valueempty1110210311优化后的状态存储格式(MapState 是 Long 类型，64 bit)历史方案的收益大大减少状态，业务反馈优化后状态仅为优化前的 20%历史方案的缺点手动改写 SQL 比较复杂，尤其一个维度有很多个枚举值或者有多个可枚举的维度输出之前需要 UDTF 进行列转行转换如何既能获得状态上的收益，又能简化 SQL 开发人员的负担？Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinD

20、istinct State 状态复用性能优化新的方案针对可枚举维度上计算 UV 的 Pattern，优化器自动进行 行转列&列转行的改写，以便复用 Distinct MapStateSELECTchannel,COUNT(DISTINCT device_id)AS uvFROM source_table-WHERE 条件指定维度的枚举值-WHERE channel in(A,B,OTHER)-WHERE channel=A OR channel=B OR channel=OTHERGROUP BY channelGroup Window Aggregate Distinct State 状态复

21、用维表 Join MiniBatch LookupJoinDistinct State 状态复用性能优化适用场景1.适用于无限流聚合和窗口聚合2.适用于单个可枚举的维度和多个可枚举的维度3.适用于简单的分组聚合和多维聚合限制条件1.至少有一个可枚举维度且枚举值可以事先确定2.当用于窗口聚合时，窗口函数必须是行语义，即不适用于 Session Window 和Dynamic Cumulate Window 等具有集合语义的窗口为什么语法上不采用 Calcite 提供的 PIVOT/UNPIVOT？Calcite 1.26.0 版本引入 PIVOT，1.27.0 引入 UNPIVOT。Flink

22、1.12+开始依赖Calcite 1.26.0 版本。Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinMiniBatch LookupJoin性能优化目标减少 RPC 调用次数原理算子攒一批数据后，一起调用维表的查询接口，执行批量查询语法在维表后面加上 hint（/*+MINI_BATCH*/）开启 MiniBatch 优化。该 Hint 有两个合法参数，一个有效的 MINI_BATCH Hint 至少包含一个参数。ALLOW_LATENCY：可选参数，Duration 类型，表示攒批时间间隔SIZE:

23、可选参数，Long 类型，必须是正数，表示每个批次最多包含的数据条目数上限Group Window Aggregate Distinct State 状态复用维表 Join MiniBatch LookupJoinMiniBatch LookupJoin性能优化案例打宽订单表，关联客户的相关信息SELECT o.order_id,o.total,c.country,c.zipFROMOrders AS oJOINCustomers/*+MINI_BATCH(ALLOW_LATENCY=5s,SIZE=10000)*/FOR SYSTEM_TIME AS OF o.proc_time AS cO

24、N o.customer_id=c.id;#4稳定性提升稳定性提升#1Group Window Aggregate 的数据倾斜升级 FLINK SQL 聚合逻辑后从状态恢复#2Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Group Window Aggregate 解决数据倾斜稳定性提升 MiniBatch Local-Global Split Distinct123Aggregate State 兼容-业务场

25、景稳定性提升日常任务指标调整，新增/删除 指标。重大活动过程中，新增天级累计指标。重大活动过程中，新增持续累计指标。123Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容-升级方式稳定性提升 有损升级，丢弃状态 业务低峰期升级，比如零点 新增任务单独计算 资源冗余（重复消费，重复计算、主备链路）天级持续Group Window Aggregate 解决数据倾斜问题Aggregat

26、e State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升if(!Arrays.equals(previousTypes,newTypes)return TypeSerializerSchemaCompatibility.incompatible();State 是否兼容判断逻辑Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State

27、 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容-目标稳定性提升降低成本随时升级增加/删除Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 尾部追加新增聚合函数新增 任意位置删除聚合函数删除 指标标识、DataType完全一致 不允许调整聚合函数的顺序 不允许同时新增和删除聚合函数兼容指标标识State DataTypesum1Lo

28、ngcount1Longavg1Long(sum)Long(count)user_define_agg_funjava_objectGroup Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 不能使用Null来填充新增聚合函数的初始值 sum的初始值是Null count的初始值是0 用户UDAF，初始值不可控 AggHandler#createAccumulators作为初始值

29、 定义state时Descriptor不能设置初始值 null 作为判断依据，Mini-batch 第一条数据初始值Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 Window early-fire/late-fire 会引入Retract消息 PreviousState比WindowState多了Window Property State 兼容判断时需要考虑该场景

30、Copy出新state时，必须使用原state中Window Property的值Retract StateGroup Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 增加指标标识和DataType对应关系 增加是否包含Window Property信息标识元信息 修改Meta 对应的Version值 根据Version决定是否读取新增元信息兼容Group Window Aggr

31、egate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 TTL Serializer 适配 根据实际State Serializer判断是否兼容 使用实际State copy数据 TTL 时间戳和原始数据保持一致TTL信息Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容

32、 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 用户学习成本低，无明显感知 不依赖外部服务，架构简单方案优点 修改Flink源码，升级版本时增加成本 应用场景限制，目前只支持Aggregate方案不足Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升凭空创建state创建State修改State任意修改state删除State任意删

33、除state增加State任意增加stateGroup Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案Aggregate State 兼容稳定性提升 不侵入Flink源码，方便Flink版本升级 Web界面化操作，无需代码开发 全场景State支持方案优点 用户学习成本高 Flink State分类，OperatorState/KeyedState Operator中是否包含State 平台开发周期，前后端支持方案不足Group Window Aggregate 解决数据倾斜问题Aggregate State 兼容 背景 兼容规则 初始值填充 Window Retract State State 向前兼容 TTL State 方案总结 终极方案#5未来展望未来展望#1数据湖、湖仓一体的落地Streaming SQL 功能扩展 提效降本 探索更多场景下的状态兼容流批一体 Batch SQL 能力增强 业务落地#2#3THANKS