[Demo] Use 3 line SQL achieve syncing MySQL to Elasticsearch

[leonardBang]

三行 SQL 实现 单品实时累计排行榜

1. 下载 docker-compose.yml 文件， 通过 docker-compose up -d 启动容器，第一次会下载镜像比较慢，之后会很快

version: '2.1'
services:
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
  elasticsearch:
    image: elastic/elasticsearch:7.6.0
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ports:
      - "9200:9200"
      - "9300:9300"
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: elastic/kibana:7.6.0
    ports:
      - "5601:5601"

2. 下载 Flink 1.14 和对应的 SQL connector jar

• 1Flink 1.14.4：https://dlcdn.apache.org/flink/flink-1.14.4/flink-1.14.4-bin-scala_2.11.tgz
• 2 MySQL-CDC ：https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.2.1/flink-sql-connector-mysql-cdc-2.2.1.jar
• 3 ElasticSearch ： https://repo.maven.apache.org/maven2/org/apache/flink/flink-sql-connector-elasticsearch7_2.11/1.14.4/flink-sql-connector-elasticsearch7_2.11-1.14.4.jar

3. 构造 MySQL 表测试数据

docker-compose exec mysql mysql -uroot -p123456

-- 创建数据库
CREATE DATABASE mydb;


USE mydb;
-- 订单表
CREATE TABLE orders (
  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  order_date DATETIME NOT NULL,
  customer_name VARCHAR(255) NOT NULL,
  price DECIMAL(10, 5) NOT NULL,
  product_id INTEGER NOT NULL,
  product_name VARCHAR(255) NOT NULL
) AUTO_INCREMENT = 10001;

INSERT INTO orders VALUES
    (default, '2022-05-20 10:08:22', 'Leonard', 13.50, 102, "Scooter"),
    (default, '2022-05-20 10:09:22', 'Leonard', 13.50, 102, "Scooter"),
    (default, '2022-05-20 10:08:29', 'Leonard', 15.00, 103, "Cat"),
    (default, '2022-05-20 10:09:49', 'Leonard', 15.00, 103, "Cat"),
    (default, '2022-05-20 10:45:30', 'Timo', 15.00, 103,  "Cat"),
    (default, '2022-05-20 10:30:30', 'Sally', 15.25, 103, "Cat"),
    (default, '2022-05-20 10:08:30', 'Edward', 25.25, 104, "Jacket"),
    (default, '2022-05-20 10:10:30', 'Edward', 25.25, 104, "Jacket"),
    (default, '2022-05-20 10:45:45', 'Timo', 25.25, 104, "Jacket"),
    (default, '2022-05-20 10:30:30', 'Sally', 15.25, 107, "Shoes"),
    (default, '2022-05-20 10:31:00', 'Kai', 13.25, 107, "Shoes"),
    (default, '2022-05-20 10:33:30', 'Sally', 15.25, 106, "Tire"),
    (default, '2022-05-20 10:15:30', 'Jark', 25.25, 106, "Tire"),
    (default, '2022-05-20 10:11:09', 'Sally', 15.00, 105, "Rocks"),
    (default, '2022-05-20 10:34:09', 'Jark', 15.00, 108, "Basket");

4. Flink CDC 实时读取MySQL数据并写入Elasticsearch

• 三行 SQL 实现实时统计单品的累计排行榜功能

-- Flink SQL                   
Flink SQL> SET execution.checkpointing.interval = 3s;

Flink SQL> -- 第一行 SQL
CREATE TABLE orders (
   order_id INT,
   order_date TIMESTAMP(3),
   customer_name STRING,
   price DECIMAL(10, 5),
   product_id INT,
   product_name STRING,
   PRIMARY KEY (order_id) NOT ENFORCED
 ) WITH (
   'connector' = 'mysql-cdc',
   'hostname' = 'localhost',
   'port' = '3306',
   'username' = 'root',
   'password' = '123456',
   'database-name' = 'mydb',
   'table-name' = 'orders'
 );

Flink SQL> -- 第二行 SQL
 CREATE TABLE top_products(
    product_name STRING,
    sales_cnt BIGINT NOT NULL,
   PRIMARY KEY (product_name) NOT ENFORCED
 ) WITH (
     'connector' = 'elasticsearch-7',
     'hosts' = 'http://localhost:9200',
     'index' = 'top_products'
 );

-- 单品累计销量统计 
Flink SQL> -- 第三行 SQL 
INSERT INTO top_products  
select product_name, count(*) as cnt from orders group by product_name;

5. 读取 Elasticsearch中的数据做可视化呈现

首先访问 http://localhost:5601/app/kibana#/management/kibana/index_pattern 创建 index： top_products.
然后就可以在 http://localhost:5601/app/kibana#/discover 看到写入的数据了.

在 MySQL 订单表中继续下单，观察 Kibana 仪表盘中数据变化：

INSERT INTO orders
VALUES (default, '2022-05-20 10:08:22', 'Leonard', 13.50, 108, "Basket");