手机网站开发企业,蜘蛛网是个什么网站,h5移动端开发,阿里大鱼 wordpress简介#xff1a; 《实时数仓入门训练营》由阿里云研究员王峰、阿里云资深技术专家金晓军、阿里云高级产品专家刘一鸣等实时计算 Flink 版和 Hologres 的多名技术/产品一线专家齐上阵#xff0c;合力搭建此次训练营的课程体系#xff0c;精心打磨课程内容#xff0c;直击当下…简介 《实时数仓入门训练营》由阿里云研究员王峰、阿里云资深技术专家金晓军、阿里云高级产品专家刘一鸣等实时计算 Flink 版和 Hologres 的多名技术/产品一线专家齐上阵合力搭建此次训练营的课程体系精心打磨课程内容直击当下同学们所遇到的痛点问题。由浅入深全方位解析实时数仓的架构、场景、以及实操应用7 门精品课程帮助你 5 天时间从小白成长为大牛本文整理自直播《实时计算 Flink 版 SQL 实践-李麟海豹》 视频链接:https://c.tb.cn/F3.0dBssY 内容简要 一、实时计算Flink版SQL简介 二、实时计算Flink版SQL上手示例 三、开发常见问题和解法 实时计算Flink版SQL简介 一关于实时计算Flink版SQL 实时计算Flink版选择了SQL这种声明式语言作为顶层API比较稳定也方便用户使用。Flink SQL具备流批统一的特性给用户统一的开发体验并且语义一致。另外Flink SQL能够自动优化包括屏蔽流计算里面State的复杂性也提供了自动优化的Plan并且还集成了AutoPilot自动调优的功能。Flink SQL的应用场景也比较广泛包括数据集成、实时报表、实时风控还有在线机器学习等场景。 二基本操作 在基本操作上可以看到SQL的语法和标准SQL非常类似。示例中包括了基本的SELECT、FILTER操作。可以使用内置函数如日期的格式化也可以使用自定义函数比如示例中的汇率转换就是一个用户自定义函数在平台上注册后就可以直接使用。 三维表 Lookup Join 在实际的数据处理过程中维表的Lookup Join也是一个比较常见的例子。 这里展示的是一个维表INNER JOIN示例。 例子中显示的SOURCE表是一个实时变化的订单信息表它通过INNER JOIN去关联维表信息这里标黄高亮的就是维表JOIN的语法可以看到它和传统的批处理有一个写法上的差异多了FOR SYSTEM_TIME AS OF这个子句来标明它是一个维表JOIN的操作。SOURCE表每来一条订单消息它都会触发维表算子去做一次对维表信息的查询所以把它叫做一个Lookup Join。 四Window Aggregation Window Aggregation窗口聚合操作也是常见的操作Flink SQL中内置支持了几种常用的Window类型比如Tumble WindowSession WindowHop Window还有新引入的Cumulate Window。 Tumble Tumble Window可以理解成固定大小的时间窗口也叫滚窗比如说5分钟、10分钟或者1个小时的固定间隔的窗口窗口之间没有重叠。 Session Session Window会话窗口 定义了一个连续事件的范围窗口定义中的一个参数叫做Session Gap表示两条数据的间隔如果超过定义的时长那么前一个Window就结束了同时生成了一个新的窗口。 Hop Hop Window不同于滚动窗口的窗口不重叠滑动窗口的窗口之间可以重叠。滑动窗口有两个参数size 和 slide。size 为窗口的大小slide 为每次滑动的步长。如果slide size则窗口会重叠同一条数据可能会被分配到多个窗口如果 slide size则等同于 Tumble Window。如果 slide size窗口之间没有重叠且有间隙。 Cumulate Cumulate Window累积窗口是Flink社区1.13版本里新引入的可以对比 Hop Window来理解区别是从Window Start开始不断去累积。示例中Window 1、Window 2、Window 3是在不断地增长的。它有一个最大的窗口长度比如我们定义Window Size是一天然后Step步长是1个小时那么它会在一天中的每个小时产生累积到当前小时的聚合结果。 看一个具体的Window聚合处理示例。 如上图所示比如说需要进行每5分钟单个用户的点击数统计。 源数据是用户的点击日志我们期望算出每5分钟单个用户的点击总数 SQL 中使用的是社区最新的 WindowTVF语法先对源表开窗再 GROUP BY 窗口对应的属性 window_start和window_end COUNT(*)就是点击数统计。 可以看到当处理12:00到12:04的数据有2个用户产生了4次点击分别能统计出来用户Mary是3次Bob是1次。在接下来一批数据里面又来了3条数据对应地更新到下一个窗口中分别是1次和2次。 五Group Aggregation 相对于Window Aggregation来说Group Aggregation直接触发计算并不需要等到窗口结束适用的一个场景是计算累积值。 上图的例子是单个用户累积到当前的点击数统计。从Query上看写法相对简单一点直接 GROUP BY user 去计算COUNT(*)就是累积计数。 可以看到在结果上和Window的输出是有差异的在与Window相同的前4条输入数据Group Aggregation输出的结果是Mary的点击数已更新到3次具体的计算过程可能是从1变成2再变成3Bob是1次随着后面3条数据的输入Bob对应的点击数又会更新成2次对结果是持续更新的过程这和Window的计算场景是有一些区别的。 之前Window窗口里面输出的数据在窗口结束后结果就不会再改变而在Group Aggregation里同一个Group Key的结果是会产生持续更新的。 六Window Aggregation Vs Group Aggregation 更全面地对比一下Window和Group Aggregation的一些区别。 Window Aggregation在输出模式上是按时输出是在定义的数据到期之后它才会输出。比如定义5分钟的窗口结果是延迟输出的比如00:00~00:05这个时间段它会等整个窗口数据都到齐之后才完整输出出来并且结果只输出一次不会再改变。 Group Aggregation是数据触发比如第一条数据来它就会输出结果同一个Key 的第二条数据来结果会更新所以在输出流的性质上两者也是不一样的。Window Aggregation一般情况下输出的是Append Stream而在Group Aggregation输出的是Update Stream。 在状态State处理上两者的差异也比较大。Window Aggregation会自动清理过期数据用户就不需要额外再去关注 State的膨胀情况。Group Aggregation是基于无限的状态去做累积所以需要用户根据自己的计算场景来定义State的TTL就是State保存多久。 比如统计一天内累计的PV和UV不考虑数据延迟的情况也至少要保证State的TTL要大于等于一天这样才能保证计算的精确性。如果State的TTL定义成半天统计值就可能不准确了。 对输出的存储要求也是由输出流的性质来决定的。在Window的输出上因为它是Append流所有的类型都是可以对接输出的。而Group Aggregatio输出了更新流所以要求目标存储支持更新可以用Hologres、MySQL或者HBase这些支持更新的存储。 实时计算 Flink 版SQL上手示例 下面通过具体的例子来看每一种SQL操作在真实的业务场景中会怎么使用比如SQL基本的语法操作包括一些常见的Aggregation的使用。 一示例场景说明电商交易数据 - 实时数仓场景 这里的例子是电商交易数据场景模拟了实时数仓里分层数据处理的情况。 在数据接入层我们模拟了电商的交易订单数据它包括了订单ID商品ID用户ID交易金额商品的叶子类目交易时间等基本信息这是一个简化的表。 示例1会从接入层到数据明细层完成一个数据清洗工作此外还会做类目信息的关联然后数据的汇总层我们会演示怎么完成分钟级的成交统计、小时级口径怎么做实时成交统计最后会介绍下在天级累积的成交场景上怎么去做准实时统计。 - 示例环境内测版 演示环境是目前内测版的实时计算Flink产品在这个平台可以直接做一站式的作业开发包括调试还有线上的运维工作。 - 接入层数据 使用 SQL DataGen Connector 生成模拟电商交易数据。 接入层数据为了方便演示简化了链路用内置的SQL DataGen Connector来模拟电商数据的产生。 这里面order_id是设计了一个自增序列Connector的参数没有完整贴出来。 DataGen Connector支持几种生成模式比如可以用Sequence产生自增序列Random模式可以模拟随机值这里根据不同的字段业务含义选择了不同的生成策略。 比如order_id是自增的商品ID是随机选取了1~10万用户ID是1~1000万交易金额用分做单位 cate_id是叶子类目ID这里共模拟100个叶子类目直接通过计算列对商品ID取余来生成订单创建时间使用当前时间模拟这样就可以在开发平台上调试而不需要去创建Kafka或者DataHub做接入层的模拟。 二示例1-1 数据清洗 - 电商交易数据-订单过滤 这是一个数据清洗的场景比如需要完成业务上的订单过滤业务方可能会对交易金额有最大最小的异常过滤比如要大于1元小于1万才保留为有效数据。 交易的创建时间是选取某个时刻之后的通过WHERE条件组合过滤就可以完成这个逻辑。 真实的业务场景可能会复杂很多下面来看下SQL如何运行。 这是使用调试模式在平台上点击运行按钮进行本地调试可以看到金额这一列被过滤订单创建时间也都是大于要求的时间值。 从这个简单的清洗场景可以看到实时和传统的批处理相比在写法上包括输出结果差异并不大流作业主要的差异是运行起来之后是长周期保持运行的而不像传统批处理处理完数据之后就结束了。 三示例1-2 类目信息关联 接下来看一下怎么做维表关联。 根据刚才接入层的订单数据因为原始数据里面是叶子类目信息在业务上需要关联类目的维度表维度表里面记录了叶子类目到一级类目的关联关系ID和名称清洗过程需要完成的目标是用原始表里面叶子类目ID去关联维表补齐一级类目的ID和Name。这里通过INNER JOIN维表的写法关联之后把维表对应的字段选出来。 和批处理的写法差异仅仅在于维表的特殊语法FOR SYSTEM_TIME AS OF。 如上所示平台上可以上传自己的数据用于调试比如这里使用了1个CSV的测试数据把100个叶子类目映射到10个一级类目上。 对应叶子类目ID的个位数就是它一级类目的ID会关联到对应的一级类目信息返回它的名称。本地调试运行优点是速度比较快可以即时看到结果。在本地调试模式中终端收到1000条数据之后会自动暂停防止结果过大而影响使用。 四示例2-1 分钟级成交统计 接下来我们来看一下基于Window的统计。 第一个场景是分钟级成交统计这是在汇总层比较常用的计算逻辑。 分钟级统计很容易想到Tumble Window每一分钟都是各算各的需要计算几个指标包括总订单数、总金额、成交商品数、成交用户数等。成交的商品数和用户数要做去重所以在写法上做了一个Distinct处理。 窗口是刚刚介绍过的Tumble Window按照订单创建时间去划一分钟的窗口然后按一级类目的维度统计每一分钟的成交情况。 - 运行模式 上图和刚才的调试模式有点区别上线之后就真正提交到集群里去运行一个作业它的输出采用了调试输出直接Print到Log里。展开作业拓扑可以看到自动开启了Local-Global的两阶段优化。 - 运行日志 - 查看调试输出结果 在运行一段时间之后通过Task里面的日志可以看到最终的输出结果。 用的是Print Sink会直接打到Log里面。在真实场景的输出上比如写到Hologres/MySQL那就需要去对应存储的数据库上查看。 可以看到输出的数据相对于数据的原始时间是存在一定滞后的。 在19:46:05的时候输出了19:45:00这一个窗口的数据延迟了5秒钟左右输出前1分钟的聚合结果。 这5秒钟实际上和定义源表时WATERMARK的设定是有关系的在声明WATERMARK时是相对gmt_create字段加了5秒的offset。这样起到的效果是当到达的最早数据是 19:46:00 时我们认为水位线是到了19:45:55这就是5秒的延迟效果来实现对乱序数据的宽容处理。 五示例2-2 小时级实时成交统计 第二个例子是做小时级实时成交统计。 如上图所示当要求实时统计直接把Tumble Window开成1小时Size的Tumble Window这样能满足实时性吗按照刚才展示的输出结果具有一定的延迟效果。因此开一个小时的窗口必须等到这一个小时的数据都收到之后在下一个小时的开始才能输出上一个小时的结果延迟在小时级别的满足不了实时性的要求。回顾之前介绍的 Group Aggregation 是可以满足实时要求的。 具体来看比如需要完成小时类目以及只算小时的两个口径统计两个统计一起做在传统批处理中常用的GROUPING SETS功能在实时Flink上也是支持的。 我们可以直接GROUP BY GROUPING SETS第一个是小时全口径第二个是类目小时的统计口径然后计算它的订单数包括总金额去重的商品数和用户数。 这种写法对结果加了空值转换处理便于查看数据就是对小时全口径的统计输出的一级类目是空的需要对它做一个空值转换处理。 上方为调试模式的运行过程可以看到Datagen生成的数据实时更新到一级类目和它对应的小时上。 这里可以看到两个不同GROUP BY的结果在一起输出中间有一列ALL是通过空值转换来的这就是全口径的统计值。本地调试相对来说比较直观和方便有兴趣的话也可以到阿里云官网申请或购买进行体验。 六示例2-3 天级累积成交准实时统计 第三个示例是天级累计成交统计业务要求是准实时比如说能够接受分钟级的更新延迟。 按照刚才Group Aggregation小时的实时统计容易联想到直接把Query改成天维度就可以实现这个需求而且实时性比较高数据触发之后可以达到秒级的更新。 回顾下之前提到的Window和Group Aggregation对于内置状态处理上的区别Window Aggregation可以实现State的自动清理Group Aggregation需要用户自己去调整 TTL。由于业务上是准实时的要求在这里可以有一个替代的方案比如用新引入的Cumulate Window做累积的Window计算天级的累积然后使用分钟级的步长可以实现每分钟更新的准实时要求。 回顾一下Cumulate Window如上所示。天级累积的话Window的最大Size是到天它的Window Step就是一分钟这样就可以表达天级的累积统计。 具体的Query如上这里使用新的TVF语法通过一个TABLE关键字把Windows的定义包含在中间然后 Cumulate Window引用输入表接着定义它的时间属性步长和size 参数。GROUP BY就是普通写法因为它有提前输出所以我们把窗口的开始时间和结束时间一起打印出来。 这个例子也通过线上运行的方式去看Log输出。 - 运行模式 可以看到它和之前Tumble Window运行的结构类似也是预聚合加上全局聚合它和Tumble Window的区别就是并不需要等到这一天数据都到齐了才输出结果。 - 运行日志 – 观察调试结果 从上方示例可以看到在20:4700的时候已经有00:0000到20:4700的结果累积还有对应的4列统计值。下一个输出就是接下来的累计窗口可以看到20:4700到20:4800就是一个累计的步长这样既满足了天级别的累计统计需求也能够满足准实时的要求。 七示例小结电商交易数据-实时数仓场景 然后我们来整体总结一下以上的示例。 在接入层到明细层的清洗处理特点是相对简单也比较明确比如业务逻辑上需要做固定的过滤条件包括维度的扩展这都是非常明确和直接的。 从明细层到汇总层例子中的分钟级统计我们是用了Tumble Window而小时级因为实时性的要求换成了Group Aggregation然后到天级累积分别展示Group Aggregation和新引入的Cumulate Window。 从汇总层的计算特点来说我们需要去关注业务上的实时性要求和数据准确性要求然后根据实际情况选择Window聚合或者Group 聚合。 这里为什么要提到数据准确性 在一开始比较Window Aggregation和Group Aggregation的时候提到Group Aggregation的实时性非常好但是它的数据准确性是依赖于State的TTL当统计的周期大于TTL那么TTL的数据可能会失真。 相反在Window Aggregation上对乱序的容忍度有一个上限比如最多接受等一分钟但在实际的业务数据中可能99%的数据能满足这样的要求还有1%的数据可能需要一个小时后才来。基于WATERMARK的处理默认它就是一个丢弃策略超过了最大的offset的这些数据就会被丢弃不纳入统计此时数据也会失去它的准确性所以这是一个相对的指标需要根据具体的业务场景做选择。 开发常见问题和解法 一开发中的常见问题 上方是实时计算真实业务接触过程中比较高频的问题。 首先是实时计算不知道该如何下手怎么开始做实时计算比如有些同学有批处理的背景然后刚开始接触Flink SQL不知道从哪开始。 另外一类问题是SQL写完了也清楚输入处理的数据量大概是什么级别但是不知道实时作业运行起来之后需要设定多大的资源 还有一类是SQL写得比较复杂这个时候要去做调试比如要查为什么计算出的数据不符合预期等类似问题许多同学反映无从下手。 作业跑起来之后如何调优这也是一个非常高频的问题。 二开发常见问题解法 1.实时计算如何下手 对于上手的问题社区有很多官方的文档也提供了一些示例大家可以从简单的例子上手慢慢了解SQL里面不同的算子在流式计算的时候会有一些什么样的特性。 此外还可以关注开发者社区实时计算 Flink 版、 ververica.cn网站、 B 站的Apache Flink 公众号等分享内容。 逐渐熟悉了SQL之后如果想应用到生产环境中去解决真实的业务问题阿里云的行业解决方案里也提供了一些典型的架构设计可以作为参考。 2.复杂作业如何调试 如果遇到千行级别的复杂SQL即使对于Flink的开发同学来也不能一目了然地把问题定位出来其实还是需要遵循由简到繁的过程可能需要借助一些调试的工具比如前面演示的平台调试功能然后做分段的验证把小段SQL局部的结果正确性调试完之后再一步一步组装起来最终让这个复杂作业能达到正确性的要求。 另外可以利用SQL语法上的特性把SQL组织得更加清晰一点。实时计算Flink产品上有一个代码结构功能可以比较方便地定位长SQL里具体的语句这都是一些辅助工具。 3.作业初始资源设置如何调优 我们有一个经验是根据输入的数据初始做小并发测试一下看它的性能如何然后再去估算。在大并发压测的时候按照需求的吞吐量逐步逼近然后拿到预期的性能配置这个是比较直接但也比较可靠的方式。 调优这一块主要是借助于作业的运行是情况我们会去关注一些重点指标比如说有没有产生数据的倾斜维表的Lookup Join需要访问外部存储有没有产生IO的瓶颈这都是影响作业性能的常见瓶颈点需要加以关注。 在实时计算Flink产品上集成了一个叫AutoPilot的功能可以理解为类似于自动驾驶在这种功能下初始资源设多少就不是一个麻烦问题了。 在产品上设定作业最大的资源限制后根据实际的数据处理量该用多少资源可以由引擎自动帮我们去调到最优状态根据负载情况来做伸缩。 原文链接 本文为阿里云原创内容未经允许不得转载。
