Apache Flink 漫谈系列(06) - 流表对偶(duality)性

实际问题

很多大数据计算产品，谈系都对用户提供了SQL API，列流比如Hive,表对 Spark, Flink等，那么SQL作为传统关系数据库的谈系查询语言，是列流应用在批查询场景的。Hive和Spark本质上都是表对Batch的计算模式(在《Apache Flink 漫谈系列 - 概述》我们介绍过Spark是Micro Batching模式)，提供SQL API很容易被人理解，谈系但是列流Flink是纯流(Native Streaming)的计算模式, 流与批在数据集和计算过程上有很大的区别，如下：

批查询场景的表对特点 - 有限数据集，一次查询返回一个计算结果就结束查询 流查询场景的谈系特点 - 无限数据集，一次查询不断修正计算结果，列流查询永远不结束

我们发现批与流的表对查询场景在数据集合和计算过程上都有很大的不同，那么基于Native Streaming模式的谈系Apache Flink为啥也能为用户提供SQL API呢?

流与批的语义关系

我们知道SQL都是作用于关系表的，在传统数据库中进行查询时候，列流SQL所要查询的表对表在触发查询时候数据是亿华云不会变化的，也就是说在查询那一刻，表是一张静态表，相当于是一个有限的批数据，这样也说明SQL是源于对批计算的查询的，那么要回答Apache Flink为啥也能为用户提供SQL API，我们首先要理解流与批在语义层面的关系。我们以一个具体示例说明，如下图：

上图展现的是一个携带时间戳和用户名的点击事件流，我们先对这些事件流进行流式统计，同时在最后的流事件上触发批计算。流计算中每接收一个数据都会触发一次计算，我们以2018/4/30 22:37:45 Mary到来那一时间切片看，无论是在流还是批上计算结果都是6。也就是说在相同的数据源，香港云服务器相同的查询逻辑下，流和批的计算结果是相同的。相同的SQL在流和批这两种模式下，最终结果是一致的，那么流与批在语义上是完全相同的。

流与表的关系

流与批在语义上是一致的，SQL是作用于表的，那么要回答Apache Flink为啥也能为用户提供SQL API的问题，就变成了流与表是否具有等价性，也就是本篇要重点介绍的为什么流表具有对偶(duality)性?如下图所示，一张表可以看做为流吗?同样流可以看做是一张表吗?如果可以需要怎样的条件和变通?

MySQL主备复制

在介绍流与表的关系之前我们先聊聊MySQL的主备复制，源码库binlog是MySQL实现主备复制的核心手段，简单来说MySQL主备复制实现分成三个步骤：

Master将改变(change logs)以二进制日志事件(binary log events)形式记录到二进制日志(binlog)中; Slave将Master的binary log events拷贝到它的中继日志(relay log); Slave重做中继日志中的事件，将改变反映到数据;

具体如下图所示:

binlog

接下来我们从binlog模式，binlog格式以及通过查看binlog的具体内容来详尽介绍binlog与表的关系。

1. binlog模式

上面介绍的MySQL主备复制的核心手段是利用binlog实现的，那边binlog会记录那些内容呢?binlog记录了数据库所有的增、删、更新等操作。MySQL支持三种方式记录binlog:

(1) statement-based logging - Events contain SQL statements that produce data changes (inserts, updates, deletes);

(2) row-based logging - Events describe changes to individual rows;

(3) mixed-base logging - 该模式默认是statement-based，当遇到如下情况会自动切换到row-based:

NDB存储引擎，DML操作以row格式记录; 使用UUID()、USER()、CURRENT_USER()、FOUND_ROWS()等不确定函数; 使用Insert Delay语句; 使用用户自定义函数(UDF); 使用临时表;

2. binlog格式

我们以row-based 模式为例介绍一下binlog的存储格式 ，所有的 binary log events都是字节序列，由两部分组成：

event header event data

关于event header和event data 的格式在数据库的不同版本略有不同，但共同的地方如下：

+=====================================+ | event  | timestamp         0 : 4    | | header +----------------------------+ |        | type_code         4 : 1    | |        +----------------------------+ |        | server_id         5 : 4    | |        +----------------------------+ |        | event_length      9 : 4    | |        +----------------------------+ |        |不同版本不一样(省略)          | +=====================================+ | event  | fixed part                 | | data   +----------------------------+ |        | variable part              | +=====================================+ 

这里有个值得我们注意的地方就是在binlog的header中有一个属性是timestamp，这个属性是标识了change发生的先后顺序，在备库进行复制时候会严格按照时间顺序进行log的重放。

3. binlog的生成

我们以对MySQL进行实际操作的方式，直观的介绍一下binlog的生成，binlog是二进制存储的，下面我们会利用工具查看binlog的文本内容。

查看一下binlog是否打开： show variables like log_bin     -> ; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | log_bin       | ON    | +---------------+-------+ 1 row in set (0.00 sec)  查看一下binlog的模式(我需要row-base模式)： show variables like binlog_format; +---------------+-------+ | Variable_name | Value | +---------------+-------+ | binlog_format | ROW   | +---------------+-------+ 1 row in set (0.00 sec)  清除现有的binlog MySQL> reset master; Query OK, 0 rows affected (0.00 sec)创建一张我们做实验的表MySQL> create table tab(     ->    id INT NOT NULL AUTO_INCREMENT,     ->    user VARCHAR(100) NOT NULL,     ->    clicks INT NOT NULL,     ->    PRIMARY KEY (id)     -> ); Query OK, 0 rows affected (0.10 sec) MySQL> show tables; +-------------------+ | Tables_in_Apache Flinkdb | +-------------------+ | tab         | +-------------------+ 1 row in set (0.00 sec)  进行DML操作 MySQL> insert into tab(user, clicks) values (Mary, 1); Query OK, 1 row affected (0.03 sec) MySQL> insert into tab(user, clicks) values (Bob, 1); Query OK, 1 row affected (0.08 sec) MySQL> update tab set clicks=2 where user=Mary     -> ; Query OK, 1 row affected (0.06 sec) Rows matched: 1  Changed: 1  Warnings: 0 MySQL> insert into tab(user, clicks) values (Llz, 1); Query OK, 1 row affected (0.08 sec) MySQL> update tab set clicks=2 where user=Bob; Query OK, 1 row affected (0.01 sec) Rows matched: 1  Changed: 1  Warnings: 0 MySQL> update tab set clicks=3 where user=Mary; Query OK, 1 row affected (0.05 sec) Rows matched: 1  Changed: 1  Warnings: 0 MySQL> select * from tab; +----+------+--------+ | id | user | clicks | +----+------+--------+ |  1 | Mary |      3 | |  2 | Bob  |      2 | |  3 | Llz  |      1 | +----+------+--------+ 3 rows in set (0.00 sec)  查看正在操作的binlog MySQL> show master status\G 

很赞哦!（844）