从 MySQL 到 ES -- 大宽表解决方案 (二)

上一篇文章里我使用了 canal 将 mysql 的数据同步到了 es 当中，但也留下了一些遗憾，像表间关联不能超过两级，只能从 mysql 中同步数据，稍复杂的逻辑就要自己实现 client，还有实践过程中不停的踩坑等等问题。所以我后来又寻找了很多其他方案， 然后发现了 2020 年才刚起步的 Flink CDC 项目，它是 Flink 的一个衍生项目，CDC 全称 change data capture，是用来进行变更数据的捕获的。看了官方文档和一些社区会议视频便爱上了这个项目，配置方便，功能又那么强大。接下来我按上次的 sku 再来演示一下。

Flink CDC

大家可能都听过 Flink， Flink 是一款分布式的计算引擎，它可以用来做批处理，即处理静态的数据集、历史的数据集；也可以用来做流处理，即实时地处理一些实时数据流，实时地产生数据的结果；也可以用来做一些基于事件的应用。

那 Flink CDC 又是什么呢，如果说 Flink 是用来处理流的， 那 Flink CDC 就是那个来提供流的， 它是 Flink 的 source。它会进行各种数据源的 change 的捕获， 然后来提供给 Flink， 之后 Flink 再进行计算， 然后 sink 到输出端。

引用一张官方的图来解释下。

左边是数据源， 右边是要将处理或者计算好的数据输出的端， 中间就是 Flink CDC 和 Flink 相结合的过程。

而且 Flink 还有一个极大的优势是， 它描述数据源和描述输出的方式， 就和 SQL 语法一样，写起来极为丝滑和舒适。

接下来先让我们像上期那样，将 Mysql 的三张表聚合打到 ES 中，记住，这只是 Flink CDC 小试身手。

复刻上期

数据准备

首先依然是 docker 内启动 mysql5.7，修改 ini 为 binlog row 模式，然后创好三张表，初始化一些数据（可以见上一篇文章）。

然后是 es 的 mappings

    "mappings": {
        "properties": {
            "id": {
                "type": "integer"
            },
            "category_id": {
                "type": "integer"
            },
            "spu_id": {
                "type": "integer"
            },
            "name": {
                "type": "keyword"
            },
            "category_name": {
                "type": "keyword"
            },
            "spu_name": {
                "type": "keyword"
            }
        }
    }

Flink CDC 配置

下载 Flink https://archive.apache.org/dist/flink/flink-1.16.0/flink-1.16.0-bin-scala_2.12.tgz

解压到 flink-1.16.0 文件夹中

然后下载 mysql 和 es 的依赖 jar 包

https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7/1.16.0/flink-sql-connector-elasticsearch7-1.16.0.jar

https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.3.0/flink-sql-connector-mysql-cdc-2.3.0.jar

直接放到放到 flink-1.16.0/lib 目录下即可。

然后到 flink-1.16.0/bin 目录下执行 ./start-cluster.sh 就运行了起来。

然后我们可以在 localhost:8081 上看到界面， 对，没错，它竟然还有界面，因为依托于强大的 Flink。

创建 CDC Source

然后就开始最神奇的一步，

我们在 flink-1.16.0/bin 执行 ./sql-client.sh

会出现一个 flink sql 的输入界面。

然后我们将我们的 source 像创表一样输入进去。

-- Flink SQL
Flink SQL> CREATE TABLE category (
    id INT,
    name STRING,
    PRIMARY KEY (id) NOT ENFORCED
  ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '654321',
    'database-name' = 'es-test',
    'table-name' = 'category'
  );

Flink SQL> CREATE TABLE spu (
   id INT,
   category_id INT,
   name STRING,
   PRIMARY KEY (id) NOT ENFORCED
 ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '654321',
    'database-name' = 'es-test',
    'table-name' = 'spu'
 );

Flink SQL> CREATE TABLE sku (
   id INT,
   spu_id INT,
   name STRING,
   PRIMARY KEY (id) NOT ENFORCED
 ) WITH (
    'connector' = 'mysql-cdc',
    'hostname' = 'localhost',
    'port' = '3306',
    'username' = 'root',
    'password' = '654321',
    'database-name' = 'es-test',
    'table-name' = 'sku'
 );

就这样我们定义好了我们的 source。

创建 CDC Sink

然后就要创建 Sink 了，Sink 就是我们要输出的地方，也像创表那样简单，这次我们输出到 es 中

Flink SQL> CREATE TABLE product (
   id INT,
   category_id INT,
   spu_id INT,
   name STRING,
   category_name STRING,
   spu_name STRING,
   PRIMARY KEY (id) NOT ENFORCED
 ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://localhost:9200',
     'index' = 'product'
 );

Source => Sink

最后就是建立对应关系

Flink SQL> INSERT INTO product
select k.id as id, c.id as category_id, p.id as spu_id, k.name as name, c.name as category_name, p.name as spu_name
from sku k 
left join spu p on p.id = k.spu_id 
left join category c on c.id = p.category_id;

对没错，它连建立对应关系都像 sql 一样，只要 select 出来，然后 insert 到要输出的表中就行。
当我们按下回车

它会告诉我们 job 已经成功的提交了。

这时候我们我们打开 kibana 其实已经可以成功的看到数据过来了

然后我们改下 category 的名称，测试一些多次关联后的表的变动是否会同步

将 category id 为 1 的 name 加了 - test 之后，刷新我们可以看到 es 内的数据也同步进行了变更，说明是会进行同步的。

Flink 流处理

打卡 Flink 的 web 界面，其实我们是可以看到 job 的处理流的

可以看到它先进行了 sku 和 spu 的计算，再和 category 进行计算，最后流打入到我们的 es 中。

踩的小坑

这次使用 Flink CDC 相比上次的 Canal 还是顺利了许多，虽然也有踩一点小坑。

insert 的表的字段顺序要和 select 的顺序一致

我一开始 select 的顺序没跟 es 一样，导致报错了，我以为会根据名称自动匹配，结果还要顺序一致。

Mysql 时区要和 Flink 配置的时区一致

insert 命令执行后我发现界面没有报错，但是 es 里一直没数据。

一番排查查找日志后我发现是 Mysql 里面的时区跟 Flink 不一致导致的

说我的 timezone 不是 Asia/Shanhai，然后我一看还真不是，我是 UTC，然后 timezone 用的 System，System 又是因为 docker 启动的，所以也是 UTC。

set global time_zone='Asia/Shanghai';

设置好之后就可以正常同步了。

支持异源

支持异源，表示 Flink CDC 可以不仅从一个库拿数据，它还可以同时从多个库拿数据，并且这些库还可以是不同的数据库架构。这也是 Canal 无法比拟的地方，Canal 只能同步 Mysql 的数据。

现在我通过一个例子来演示一下，Flink CDC 的这个能力。

我们现在已经有了 category，spu，sku 表了。

此时我们又使用 mongoDB 存储了 sku 的素材数据，因为 mongo 比较灵活，所以我们使用 mongo 来存这个数据。

然后我们又利用 postgress 的性能高效，用它来存储了 sku 的库存数据。

这时候如果我们再需要把这些数据同步到 es 里，对以前来说会比较困难，但对 Flink 来说就像之前创表再 insert 一样容易。

然后我们配置一下 mongo 和 postgress 的环境和数据

mongoDB

像之前一样，我们先在 docker 内把 mongoDB 跑起来

docker run --name mongo \
-v /home/robinson/code/mongo_data:/data/db \
-p 0.0.0.0:27017:27017 \
-e MONGO_INITDB_ROOT_USERNAME=root \
docker.io/library/mongo:5.0 \
--replSet rs0

创库创集合填充数据

// 创表
use data
// 创一个collection
db.createCollection("material")
// 然后往collection内预置数据
db.material.insertMany([
{
    "sku_id" : 1,
    "detail" : [
    {
            "material_type" : "1",
            "material_explain" : "主图",
            "content" : "https://testtest.com/img.jpg"
    },
        {
             "material_type" : "2",
            "material_explain" : "详情页图",
            "content" : "https://testtest.com/img2.jpg"
    }
    ]
},
{
    "sku_id" : 2,
    "detail" : [
    {
            "material_type" : "1",
            "material_explain" : "主图",
            "content" : "https://testtest.com/img3.jpg"
    },
        {
             "material_type" : "2",
            "material_explain" : "详情页图",
            "content" : "https://testtest.com/img4.jpg"
    }
    ]
},
{
    "sku_id" : 3,
    "detail" : [
    {
            "material_type" : "1",
            "material_explain" : "主图",
            "content" : "https://testtest.com/img5.jpg"
    },
        {
             "material_type" : "2",
            "material_explain" : "详情页图",
            "content" : "https://testtest.com/img6.jpg"
    }
    ]
}
])

创建用来给 flink 读流的角色

use admin;
db.createRole(
    {
        role: "flinkrole",
        privileges: [{
            // Grant privileges on all non-system collections in all databases
            resource: { db: "", collection: "" },
            actions: [
                "splitVector",
                "listDatabases",
                "listCollections",
                "collStats",
                "find",
                "changeStream" ]
        }],
        roles: [
            // Read config.collections and config.chunks
            // for sharded cluster snapshot splitting.
            { role: 'read', db: 'config' }
        ]
    }
);

postgress

同样先是用 docker 跑一个 postgress 出来

 docker run --name postgres \
-e POSTGRES_PASSWORD=654321 \
-v /home/robinson/code/postgres_data:/var/lib/postgresql/data \
-p 0.0.0.0:5432:5432 \
docker.io/library/postgres:12

然后创表填充数据

create table stock
(
    id        serial,
    sku_id    integer,
    stock_num integer
);
INSERT INTO supply.stock (id, sku_id, stock_num) VALUES (1, 1, 123123);
INSERT INTO supply.stock (id, sku_id, stock_num) VALUES (2, 2, 342);

在 Flink 中 create

准备工作做完后就是像 mysql 一样 在 Flink Sql 内创表

material 表

CREATE TABLE material (
   _id STRING,
   sku_id INT,
   detail ARRAY<ROW<material_type STRING, material_explain STRING, content STRING>>,
   PRIMARY KEY (_id) NOT ENFORCED
 ) WITH (
   'connector' = 'mongodb-cdc',
   'hosts' = 'localhost:27017',
   'username' = 'flink',
   'password' = '654321',
   'database' = 'data',
   'collection' = 'material'
 );

stock 表

CREATE TABLE stock (
  id INT,
  sku_id INT,
  stock_num INT,
  PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'localhost',
  'port' = '5432',
  'username' = 'postgres',
  'password' = '654321',
  'database-name' = 'postgres',
  'schema-name' = 'supply',
  'table-name' = 'stock'
);

创建新的 es 的 mapping

PUT /product_v2
{
    "mappings": {
        "properties": {
            "id": {
                "type": "integer"
            },
             "category_id": {
                "type": "integer"
            },
             "spu_id": {
                "type": "integer"
            },
            "name": {
                "type": "keyword"
            },
            "category_name": {
                "type": "keyword"
            },
            "spu_name": {
                "type": "keyword"
            },
            "material": {
                "type": "object"
            },
            "stock_num": {
                "type": "integer"
            }
        }
    }
}

然后在 Flink 里再创这个新的 product_v2 的表

CREATE TABLE product_v2 (
   id INT,
   category_id INT,
   spu_id INT,
   name STRING,
   category_name STRING,
   spu_name STRING,
   material ARRAY<ROW<material_type STRING, material_explain STRING, content STRING>>,
   stock_num INT,
   PRIMARY KEY (id) NOT ENFORCED
 ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://localhost:9200',
     'index' = 'product_v2'
 );

然后就是写最后的 insert 预计，将表聚合导到 es 中

INSERT INTO product_v2
select k.id as id, c.id as category_id, p.id as spu_id, k.name as name, c.name as category_name, p.name as spu_name, m.detail as material, s.stock_num as stock_num
from sku k 
left join spu p on p.id = k.spu_id 
left join category c on c.id = p.category_id
left join material m on m.sku_id = k.id
left join stock s on s.sku_id = k.id;

回车

然后我们可以看到历史数据已经同步到了 es 中了，素材和库存也同步过来了。

然后我们再改下 mongoDB 里面的素材图片地址

可以看到 es 内的图片地址也随之发生了变更

然后我们到 Flink 的 web 界面查看，可以看到它整个的计算过程。

多端输出

我之前设置的 sink 只有 ES，所以数据只输出到了 es 中，那如果有一天其他组的同学说要接收 sku 变更的事件怎么办。

我们不用在代码里加写入事件的逻辑，可以直接写一个 Sink， 将输出写入到 Kafka 中就行。

接下来我演示一下。

kafka 配置

因为启动 kafka 的同时还要启动 zookeeper，所以这次放到了 docker-compose-file 里面

version: '3'

services:
  zookeeper:
    image: docker.io/wurstmeister/zookeeper
    restart: unless-stopped
    ports:
      - "0.0.0.0:2181:2181"
    # volumes:
    #   - /usr/local/zookeeper/data:/data
    #   - /usr/local/zookeeper/log:/datalog
    container_name: zookeeper

  kafka:
    image: docker.io/wurstmeister/kafka
    ports:
      - "0.0.0.0:9092:9092"
    environment:
      KAFKA_ADVERTISED_HOST_NAME: "172.23.80.104"
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      # KAFKA_LOG_DIRS: "/kafka/kafka-logs-1"
    # volumes:
    #   - /usr/local/kafka/logs:/kafka/kafka-logs-1
    depends_on:
      - zookeeper
    container_name: kafka

创建 topic 测试

docker exec -it kafka kafka-console-producer.sh --broker-list 172.23.80.104:9092 --topic test
docker exec -it kafka kafka-console-consumer.sh --bootstrap-server 172.23.80.104:9092 --topic test --from-beginning

是可以通的。

然后就是像之前那样在 Flink 中创表

CREATE TABLE KafkaTable8 (
   id INT,
   category_id INT,
   spu_id INT,
   name STRING,
   category_name STRING,
   spu_name STRING,
   PRIMARY KEY (id) NOT ENFORCED
) WITH (
  'connector' = 'upsert-kafka',
  'topic' = 'test',
  'properties.bootstrap.servers' = '172.23.80.104:9092',
  'key.format' = 'json',
  'value.format' = 'json'
);

然后写 insert 语句创建连接

INSERT INTO KafkaTable8
select k.id as id, c.id as category_id, p.id as spu_id, k.name as name, c.name as category_name, p.name as spu_name
from sku k 
left join spu p on p.id = k.spu_id 
left join category c on c.id = p.category_id;

然后我们就可以看到 kafka 中会有事件同步过来了。

当然我们也可以像事件一样将数据 Sink 到 Redis 中，这样连缓存的建立都可以和业务代码解耦，不用再在业务中写一套写入和更新缓存的操作，所以未来 Flink 有无尽的可能等待着去探索。

对了，这次研究这个 Flink CDC 项目还给官方仓库提了一个 PR 被 Merge 了，虽然只是教程的一些错误，但还是很开心，下次争取源码的 PR。