Apache Flink 漫谈系列 - 一连查询(Continuous Queries)

上面查询SQL的代码布局如下(这个图示在Alibaba 企业版Flink的集成IDE情形天生的，相识更多)：

上面SQL中我们发明有两层查询计较逻辑，第一个查询计较逻辑是与SOURCE相连的按地域统计订单数目的分组统计，第二个查询计较逻辑是在第一个查询产出的动态表上面举办按订单数目统计地域数目的分组统计，我们一层一层说明。

5. 错误处理赏罚

• 第一层说明：SELECT region, count(id) AS order_cnt FROM order_tab GROUP BY region;



• 第二层说明：SELECT order_cnt, count(region) as region_cnt FROM order_count_view GROUP BY order_cnt;

凭证第一层说明的功效，再说明第二层产出的功效，我们说明的进程是对的，可是最终写到sink表的计较功效是错误的，那我们错在那边了呢?

着实当 (SH,2)这笔记录来的时辰，早年来过的(SH, 1)已经是脏数据了，当(BJ, 2)来的时辰，已经参加过计较的(BJ, 1)也酿成脏数据了，同样当(BJ, 3)来的时辰，(BJ, 2)也是脏数据了，上面的说明，没有处理赏罚脏数据进而导致最终功效的错误。那么Apache Flink内部是怎样正确处理赏罚的呢?

6. 正确处理赏罚

• 第一层说明：SELECT region, count(id) AS order_cnt FROM order_tab GROUP BY region;



• 第二层说明：SELECT order_cnt, count(region) as region_cnt FROM order_count_view GROUP BY order_cnt;

上面我们将有更新的变乱举办打标的方法来处理赏罚脏数据，这样在Apache Flink内部计较的时辰 算子会按照变乱的打标来处理赏罚变乱，在aggregate function中有两个对应的要领(retract和accumulate)来处理赏罚差异标识的变乱,如上面用到的count AGG，内部实现如下：

def accumulate(acc: CountAccumulator): Unit = { 
acc.f0 += 1L // acc.f0 存储记数 
} 
 
def retract(acc: CountAccumulator, value: Any): Unit = { 
if (value != null) { 
acc.f0 -= 1L //acc.f0 存储记数 
}} 

Apache Flink内部这种为变乱举办打标的机制叫做 retraction。retraction机制保障了在流上已经流转到下流的脏数据必要被撤回题目，进而保障了一连查询的正确语义。

八、Apache Flink Connector 范例

本篇一开始就比拟了MySQL的数据存储和Apache Flink数据存储的区别，Apache Flink今朝是一个计较平台，将数据的存储以高度抽象的插件机制与各类已有的数据存储无缝对接。今朝Apache Flink中将数据插件称之为链接器Connector，Connnector又按数据的读和写分成Soruce(读)和Sink(写)两种范例。对付传统数据库表，PK是一个很重要的属性，在频仍的按某些字段(PK)举办更新的场景,在表上界说PK很是重要。那么作为完全支持ANSI-SQL的Apache Flink平台在Connector上面是否也支持PK的界说呢?

1. Apache Flink Source

此刻(2018.11.5)Apache Flink顶用于数据流驱动的Source Connector上面无法界说PK，这样在某些营业场景下会造成数据量较大，造成计较资源不须要的挥霍，乃至有聚合功效不是用户“祈望”的环境。我们以双流JOIN为例来声名：

SQL： 
 
CREATE TABLE inventory_tab( 
product_id VARCHAR, 
product_count BIGINT 
); 
 
CREATE TABLE sales_tab( 
product_id VARCHAR, 
sales_count BIGINT 
) ; 
 
CREATE TABLE join_sink( 
product_id VARCHAR, 
product_count BIGINT, 
sales_count BIGINT, 
PRIMARY KEY(product_id) 
); 
 
CREATE VIEW join_view AS 
SELECT 
l.product_id, 
l.product_count, 
r.sales_count 
FROM inventory_tab l 
JOIN sales_tab r 
ON l.product_id = r.product_id; 
 
INSERT INTO join_sink 
SELECT 
product_id, 
product_count, 
sales_count 
FROM join_view ; 

（编辑：湖南网）

【声明】本站内容均来自网络，其相关言论仅代表作者个人观点，不代表本站立场。若无意侵犯到您的权利，请及时与联系站长删除相关内容!