从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。

1	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测试

1
2

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。