阿里云体验有奖：使用PolarDB-X与Flink搭建实时数据大屏 ...

体验简介

场景将提供一台配置了CentOS 8.5操作系统的ECS实例（云服务器）。通过本教程的操作带您体验如何使用PolarDB-X与Flink搭建一个实时数据链路，模拟阿里巴巴双十一GMV大屏。

实验准备

1. 创建实验资源
开始实验之前，您需要先创建ECS实例资源。

• 在实验室页面，单击创建资源。
  
• （可选）在实验室页面左侧导航栏中，单击云产品资源列表，可查看本次实验资源相关信息（例如IP地址、用户信息等）。
  

说明：资源创建过程需要1~3分钟。
2. 安装PolarDB-X
本步骤将指导您如何安装PolarDB-X。

• 安装并启动Docker。
  
  
  
  • 执行如下命令，安装Docker。
    
  
  

1. curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun

复制代码

• 执行如下命令，启动Docker。
  

1. systemctl start docker

复制代码

• 执行如下命令，安装PolarDB-X。
  

1. docker run -d --name some-polardb-x -p 8527:8527 polardbx/polardb-x:2.1.0

复制代码

3. 在PolarDB-X中准备订单表
PolarDB-X支持通过MySQL Client命令行、第三方客户端以及符合MySQL交互协议的第三方程序代码进行连接。本实验使用MySQL Client命令行连接到PolarDB-X数据库。
本步骤将指导您如何连接PolarDB-X数据库，并创建测试库、测试表和测试数据。

• 执行如下命令，安装MySQL。
  

1. yum install mysql -y

复制代码

• 执行如下命令，查看MySQL版本号。
  

1. mysql -V

复制代码

返回结果如下，表示您已成功安装MySQL。

• 执行如下命令，登录PolarDB-X数据库。
  

说明：

• 本实验场景中的PolarDB-X数据库用户名和密码已预设，请您使用下方命令登录即可。
  
• 如遇到mysql: [Warning] Using a password on the command line interface can be insecure.ERROR 2013 (HY000): Lost connection to MySQL server at 'reading initial communication packet', system error: 0报错，请您稍等一分钟，重新执行登录命令即可。
  

1. mysql -h127.0.0.1 -P8527 -upolardbx_root -p123456

复制代码

返回结果如下，表示您已成功登录PolarDB-X数据库。

• 执行如下SQL语句，创建测试库mydb。
  

1. create database mydb;

复制代码

• 执行如下SQL语句，使用测试库mydb。
  

1. use mydb;

复制代码

• 执行如下SQL语句，创建订单表orders。
  

1. CREATE TABLE `orders` (
2. `order_id` int(11) NOT NULL AUTO_INCREMENT,
3. `order_date` datetime NOT NULL,
4. `customer_name` varchar(255) NOT NULL,
5. `price` decimal(10, 5) NOT NULL,
6. `product_id` int(11) NOT NULL,
7. `order_status` tinyint(1) NOT NULL,
8. PRIMARY KEY (`order_id`)
9. )AUTO_INCREMENT = 10001;

复制代码

• 执行如下SQL语句，给订单表orders中插入数据。
  

1. INSERT INTO orders
2. VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
3.        (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
4.        (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

复制代码

• 输入exit退出数据库。
  

4. 运行Flink
本步骤将指导您如何下载并运行Flink。

• 安装JDK。
  

a.执行如下命令，使用yum安装JDK 1.8。

1. yum -y install java-1.8.0-openjdk*

复制代码

b.执行如下命令，查看是否安装成功。

1. java -version

复制代码

返回结果如下，表示您已成功安装JDK 1.8。

• 下载Flink和Flink CDC MySQL Connector。
  

a.执行如下命令，下载Flink。

1. wget https://labfileapp.oss-cn-hangzhou.aliyuncs.com/PolarDB-X/flink-1.13.6-bin-scala_2.11.tgz

复制代码

b.执行如下命令，解压Flink。

1. tar xzvf flink-1.13.6-bin-scala_2.11.tgz

复制代码

c.执行如下命令，进入lib目录。

1. cd flink-1.13.6/lib/

复制代码

d.执行如下命令，下载flink-sql-connector-mysql-cdc。

1. wget https://labfileapp.oss-cn-hangzhou.aliyuncs.com/PolarDB-X/flink-sql-connector-mysql-cdc-2.2.1.jar

复制代码

e.执行如下命令，返回Flink目录。

1. cd ..

复制代码

• 启动Flink。
  

a.执行如下命令，启动Flink。

1. ./bin/start-cluster.sh

复制代码

b.执行如下命令，连接Flink。

1. ./bin/sql-client.sh

复制代码

• 在Flink中创建与PolarDB-X关联的订单表orders。
  

a.执行如下SQL语句，创建订单表orders。

1. CREATE TABLE orders (
2. order_id INT,
3. order_date TIMESTAMP(0),
4. customer_name STRING,
5. price DECIMAL(10, 5),
6. product_id INT,
7. order_status BOOLEAN,
8. PRIMARY KEY (order_id) NOT ENFORCED
9. ) WITH (
10. 'connector' = 'mysql-cdc',
11. 'hostname' = 'localhost',
12. 'port' = '8527',
13. 'username' = 'polardbx_root',
14. 'password' = '123456',
15. 'database-name' = 'mydb',
16. 'table-name' = 'orders'
17. );

复制代码

b.执行如下SQL语句，查看订单表orders。

1. select * from orders;

复制代码

返回结果如下，您可以查看到PolarDB-X的订单表orders的数据已经同步到Flink的订单表orders中。

c.按q键退出。
5. 启动压测脚本并实时获取GMV
经过前面几步操作后，我们在PolarDB-X中准备好了原始订单表，在Flink中准备好了对应的订单表，并通过 PolarDB-X Global Binlog与Flink CDC MySQL Connector打通了两者之间的实时同步链路。 本步骤将指导您如何创建压测脚本，模拟双十一零点大量订单涌入的场景。

• 准备压测脚本。
  

a.在实验页面，单击右上角的图标，创建新的终端二。

b.执行如下命令，创建配置文件mysql-config.cnf。

1. vim mysql-config.cnf

复制代码

c.将如下代码添加到配置文件mysql-config.cnf中。

1. [client]
2. user = "polardbx_root"
3. password = "123456"
4. host = 127.0.0.1
5. port = 8527

复制代码

d.添加完成后的文件内容如下所示。按下Esc键后，输入:wq后按下Enter键保存并退出。
e.执行如下命令，创建脚本buy.sh。

1. vim buy.sh

复制代码

f.将如下代码添加到脚本buy.sh中。

1. #!/bin/bash
3. echo "start buying..."
5. count=0
6. while :
7. do
8. mysql --defaults-extra-file=./mysql-config.cnf -Dmydb -e "insert into orders values(default, now(), 'free6om', 1024, 102, 0)"
9. let count++
10. if ! (( count % 10 )); then
11.   let "batch = count/10"
12.   echo $batch": got 10 products, gave 1024￥"
13. fi
14. sleep 0.05
15. done

复制代码

g.添加完成后的文件内容如下所示。按下Esc键后，输入:wq后按下Enter键保存并退出。
h.执行如下命令，为脚本buy.sh增加执行权限。

1. chmod +x buy.sh

复制代码

• 启动Flink实时计算。
  

本实验场景通过Flink SQL实时呈现GMV计算结果。
切换至终端一，在Flink中执行如下SQL语句，查询GMV（gmv列）和订单数（orders列）。

1. select 1, sum(price) as gmv, count(order_id) as orders from orders;

复制代码

返回结果如下，您可在Flink的实时计算结果中查看到实时的GMV（gmv列）和订单数（orders列）。

• 启动压测脚本。
  

a.切换至终端二，执行如下命令，启动压测脚本，开始创建订单。

1. ./buy.sh

复制代码

返回结果如下，您可看到压测脚本启动后，不断有订单被创建出来。

b.切换至终端一，在Flink的实时计算结果中，可查看到实时的GMV（gmv列）和订单数（orders列）。

恭喜完成

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！