「我的数据空间」实时计算实践笔记 · Flink SQL 系列引言无论在OLAP领域还是OLTP领域多表Join都是业务所必备的。在OLTP场景中日常事务的处理需要使用到Join操作OLAP场景由于数据量大、字段多数据通常被分为事实表和维度表以星型或雪花模型组成那么数据查询中多表Join的操作更是不能少。对于离线计算而言经过数据库领域多年的积累Join 语义以及实现已经十分成熟然而对于近年来刚兴起的实时领域 Streaming SQL 来说 Join 却处于刚起步的状态。Join的本质是分别从N(N 1)张表中获取不同字段进行拼接从而获得完整的数据。目前SQL中Join的种类可以分为CROSS Join交叉连接计算表的笛卡尔积。INNER Join内连接计算表的交集。OUTER JoinLEFT返回左表所有行右表不存在补NULL。RIGHT返回右表的所有行左表不存在补NULL。FULL返回左右表的并集不存在的补NULL。SELF Join自连接。在Batch SQL的模式中表中的数据是一个有限集Join的实现会依赖数据集的缓存。然而在Streaming SQL的模式中数据集是无限的。那么如何在无限的数据流中实现表的Join操作使得特定时刻的输出结果与Batch模式结果相同想要解决这个问题首先需要先了解动态表与连续查询的概念。动态表与连续查询在Flink的官方文档中专门有一页介绍了动态表与连续查询的概念在此不赘述只是简单介绍下相关概念以及为什么无限流数据可以进行查询和Join。这里需要提到一个概念——流表对偶(duality)性。对偶性是描述导致相同的物理结果表面上不同的理论之间的对应关系。那么流表对偶性即描述数据流和数据表虽然属于不同的理论但他们承载的SQL却能够通过联系拥有相同的物理结果。那么如何关联流和表呢下面的例子就能够很好地展示。例通过MySQL主从复制机制了解流表对偶性。了解MySQL的同学应该都知道Binlog是MySQL实现主从复制的关键Binlog会记录CREATE、ALTER TABLE等和INSERT、UPDATE等操作。如果选择模式为row-based那么对表的操作会以数据行为单位进行记录。记录在BInlog中的数据会发给从服务器并在从服务器上进行表的还原。根据上图理解动态表会很容易。动态表就是会随时间所变化的表随着数据地不断流入表中字段的内容也会发生变化。相对于静态表而言针对动态表的查询略有区别。针对动态表的查询并不会终止而是一直在进行类似于数据库中存在的物化视图。在进行连续查询时Flink首先会将changelog stream(INSERT、UPDATE、DELETE操作)转化为一张动态表并在此动态表上进行持续查询。再将查询的结果再生成一张结果动态表并依据此结果动态表将数据一条一条的地写入下游。双流Join了解了上文的两个概念后理解Streaming模式下的Join就会变得很简单。Flink在双流Join这一类中支持两种Join方式分别是Regular Join和Interval Join。其原理基本相同Regular Join可以认为就是Batch模式Join的另一种表现形式在同一时刻其结果与Batch模式完全相同。Interval Join为了解决Regular Join的数据无限增长的问题引入了时间窗口概念。Regular Join该模式是最基本的Join模式对数据的更新或改变对Join两边都是可见的且随时间变化能够影响全局结果。有的同学可能会有疑问流式Join数据是一条一条被处理的该条数据影响之后的Join结果很容易理解那么如何影响之前的Join结果呢看完下面的内容你可能就会有答案。上图清晰地展示了Join的处理流程。来自两边的数据(L-Event、R-Event)进入Join算子后先会被分别更新到L-State和R-State接着L-Event会和R-State中的结果进行Join操作输出Join结果发到下游同样R-Event则会去和L-State进行Join并把相关结果发往下游。假设我们有两张source表分别是Orders表和Shipments表数据源都是changelog。Orders------------------id|price|------------------01|10|------------------02|20|------------------03|30|---------------------------------------------------------------------Shipments------------------id|type|------------------01|AA|------------------03|AB|------------------05|BB|------------------通过id来进行JoinDML如下selecto.id,o.price,s.typefromOrders oLEFTJoinShipments sono.ids.id;我们假设两张表中数据的流入顺序为:Orders | Shipments ------------------------------ 1. (01, 10) | ------------------------------ 2. (02, 20) | ------------------------------ 3. | (01, AA) ------------------------------ 4. | (03, AB) ------------------------------ 5. (03, 30) | ------------------------------ 6. | (05, BB) 注号代表此数据为Insert那么Join后流出的数据为isINSERT|id|price|type-------------------------------------------1.true|01|10|NULL-------------------------------------------2.true|02|20|NULL-------------------------------------------3.false|01|10|NULLtrue|01|10|AA-------------------------------------------4.-------------------------------------------5.true|03|30|AB-------------------------------------------6.上面的情况较为简单都是append模式那么如果存在retract的数据情况又会变得复杂很多。假设在上述结果之后我们有第7条数据进入Orders | Shipments ------------------------------ 7. | - 01, AA 注- 代表delete那么对应第7条输出为:isINSERT|id|price|type-------------------------------------------7.false|01|10|AAtrue|01|10|NULL从上面的逻辑不难看出因为有着retract的存在使得Flink能够及时纠正Join结果使结果与batch模式保持一致。双流Join的具体逻辑在flink-table-runtime-blink 模块下的StreamingJoinOperator类中感兴趣的同学可以深入研究。目前Flink支持INNER/LEFT/RIGHT/FULL Join。SEMI Join AND ANTI Join在正常的Join外还存在着一类较为特殊的Join方式分别是SEMI Join和ANTI Join这两种Join的特殊之处在于他们只返回左表的列数据并不将右表的数据做输出只是用右表数据对左表进行过滤。下图直观地展现了 SEMI Join 和 ANTI Join 输出结果的异同。-- SEMI JoinSELECT*FROMEmployeeWHEREDeptNameIN(SELECTDeptNamefromDept);-- ANTI JoinSELECT*FROMEmployeeWHEREDeptNameNOTIN(SELECTDeptNamefromDept);虽然Regular Join能够像Batch模式那样满足我们的需求但是他也有着致命缺点即两张表的数据都需要缓存在state中在unbound数据的情况下占用的资源会无限增长。为了解决这样的问题且保证Join的核心逻辑Flink引入了Interval Join。Interval Join为了解决Regular Join数据持续增长的问题Flink在Interval Join中引入了时间窗口的概念窗口外的数据会被Flink清理这极大缓解了资源的占用。Interval Join的时间语义既可以是Event Time也可以是Processing TimeFlink会根据选择的时间语义来维护窗口。Orders------------------------------|order_time|id|price|------------------------------|XXXXXXXXXX|01|10|------------------------------|XXXXXXXXXX|02|20|------------------------------|XXXXXXXXXX|03|30|---------------------------------------------------------------------------------Shipments------------------------------|ship_time|id|type|------------------------------|XXXXXXXXX|01|AA|------------------------------|XXXXXXXXX|03|AB|------------------------------|XXXXXXXXX|05|BB|------------------------------例如上面的两张表如果我们需要在4小时窗口内查看商品的发货信息可以用下文DML实现CREATETABLEOrders(order_timeBIGINT,idINT,priceDOUBLE,WATERMARKFORToTIMESTAMP(order_time)asevent_time-INTERVAL60SECOND)with(..........)CREATETABLEShipments(ship_timeBIGINT,idINT,typeVARCHAR,WATERMARKFORToTIMESTAMP(ship_time)asevent_time-INTERVAL60SECOND)with(..........)INSERTINTOprintSELECT*FROMOrders o,Shipments sWHERE0.ids.idANDs.ship_timeBETWEENo.order_timeANDo.order_timeINTERVAL4HOUR;从上图可以看出根据Shipments表的watermarkorder_time小于ship_time - 4 hour的数据可以被丢弃根据Orders表的watermarkship_time小于order_time的数据会被丢弃Flink会依据watermark来进行过期数据的清理将空间维持在合适的范围。维表Join虽然Interval Join能够解决资源问题但是也给表绑定了时间界超出时间界限的数据需要被丢弃。为了支持数据量不大变化不频繁表的一类Join场景Flink引入了两种维表Join模式分别是Temporal Table Join和Temporal Table Function Join。这两种Join模式的主要应用场景是为了补全事实表probe table数据的额外字段通常维度表build table里的字段很少发生变化。从数据量上来看维度表一般数据量较小。通常维表Join的逻辑是基于Hash Join来实现的Flink中的维表Join逻辑也不例外。Hash Join的Join逻辑正常会分为两步第一步将维度表按Key散列建立哈希表第二步用事实表中的RowKey去哈希表中探测再输出结果所以通常又会将事实表叫做probe table维度表叫做build table。Temporal Table Join按维表加载方式实现逻辑主要分为两类。第一类是全量加载例如HDFSFlink会将HDFS文件全量加载进内存进行Join操作时再去内存的cache匹配第二类为部分加载例如JDBC、Hbase、Redis等会根据probe table当前Row数据的Key去数据库查询再依据情况辅之以缓存逻辑。在使用该模式进行Join操作时需要先在probe table中指定Processing time列具体的Join写法如下insertintoSinkselecto.amount,o.currency,l.rate,o.amount*l.ratefromOrderojoinLatestRateFORSYSTEM_TIMEASOFo.proctimeaslONo.currencyl.currency;需要注意的是目前Temporal Table Join仅支持INNER JOIN和LEFT JOIN。Temporal Table Join的缺点是不能指定Event time作为时间语义只支持Processing time那就是说无论probe table处在哪个时间段都会和build table中最新的数据进行Join。Temporal Table Function JoinTemporal Table Function模式主要是通过UDTF来实现probe流和Temporal table的Join。需要注意的是这里的left input(probe table)需要是append-only tableright input(build table)需要有主键和用于版本化的字段(通常是时间字段)。在具体实现逻辑中Flink会将左表数据和右表数据按Key分别保存到leftState和rightState中格式都是MapStateLong, BaseRow。左右state不同的是leftState的主键是一个递增序列rightState则以时间列作为主键。在Join逻辑中首先会遍历左边的状态state并提取元素中的时间列用时间列去排好序的rightState中进行binary search查找rightState.rowTime leftState.rowTime的第一条数据匹配上就发往下游同时在leftState状态中清除。从上面的匹配逻辑可以看出目前Temporal Table Function Join仅支持INNER Join。Temporal Table Function Join模式在SQL语言方面仅支持probe table的DDL和Join逻辑的DML暂时还不支持以SQL来创建Temporal Table Function。想要使用该模式首先需要借助Table API创建Temporal Table Function再写DML进行Join。总结实时领域的Streaming Join因为不能像Batch Join那样缓存完整数据集所以需要给缓存设定基于时间的清理机制和限定Join涉及的数据范围。Flink SQL针对这些差异从双流Join和维表Join两个方向设计推出了多种Join模式来满足日常业务需求。本文收录于「我的数据空间」技术库——一套可私有化部署的数据平台(数据集成 / 实时计算 / 数据湖 / 湖仓查询 / 智能问数)。产品介绍见[我的数据空间官网]。