使用PolarDB-X与Flink搭建实时数据大屏

本文涉及的产品
云数据库 RDS MySQL,集群系列 2核4GB
推荐场景:
搭建个人博客
云原生数据库 PolarDB 分布式版,标准版 2核8GB
实时计算 Flink 版,5000CU*H 3个月
简介: 场景将提供一台配置了CentOS 8.5操作系统的ECS实例(云服务器)。通过本教程的操作带您体验如何使用PolarDB-X与Flink搭建一个实时数据链路,模拟阿里巴巴双十一GMV大屏。

1、安装PolarDB-X

本步骤将指导您如何安装PolarDB-X。

  1. 安装并启动Docker。
  1. 执行如下命令,安装Docker。
curl -fsSL https://get.docker.com | bash -s docker --mirror Aliyun

b.执行如下命令,启动Docker。

systemctl start docker

2.执行如下命令,安装PolarDB-X。

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

2、在PolarDB-X中准备订单表

PolarDB-X支持通过MySQL Client命令行、第三方客户端以及符合MySQL交互协议的第三方程序代码进行连接。本实验使用MySQL Client命令行连接到PolarDB-X数据库。

本步骤将指导您如何连接PolarDB-X数据库,并创建测试库、测试表和测试数据。

  1. 执行如下命令,安装MySQL。
yum install mysql -y

2.执行如下命令,查看MySQL版本号。

mysql -V

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

  1. 执行如下命令,登录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报错,请您稍等一分钟,重新执行登录命令即可。
mysql -h127.0.0.1 -P8527 -upolardbx_root -p123456

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

4.执行如下SQL语句,创建测试库mydb。使用测试库mydb。

create database mydb;
use mydb;

5.执行如下SQL语句,创建订单表orders。给订单表orders中插入数据。

CREATE TABLE `orders` (
 `order_id` int(11) NOT NULL AUTO_INCREMENT,
 `order_date` datetime NOT NULL,
 `customer_name` varchar(255) NOT NULL,
 `price` decimal(10, 5) NOT NULL,
 `product_id` int(11) NOT NULL,
 `order_status` tinyint(1) NOT NULL,
 PRIMARY KEY (`order_id`)
)AUTO_INCREMENT = 10001;
INSERT INTO orders
VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

6.输入exit退出数据库。

3、运行Flink

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

  1. 安装JDK。

a.执行如下命令,使用yum安装JDK 1.8。查看是否安装成功。

yum -y install java-1.8.0-openjdk*
java -version

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

2.下载Flink和Flink CDC MySQL Connector。

a.执行如下命令,下载Flink。解压Flink。进入lib目录。下载flink-sql-connector-mysql-cdc。返回Flink目录。

wget https://labfileapp.oss-cn-hangzhou.aliyuncs.com/PolarDB-X/flink-1.13.6-bin-scala_2.11.tgz
tar xzvf flink-1.13.6-bin-scala_2.11.tgz
cd flink-1.13.6/lib/
wget https://labfileapp.oss-cn-hangzhou.aliyuncs.com/PolarDB-X/flink-sql-connector-mysql-cdc-2.2.1.jar
cd ..

3.启动Flink。

a.执行如下命令,启动Flink。连接Flink。

./bin/start-cluster.sh
./bin/sql-client.sh

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

a.执行如下SQL语句,创建订单表orders。查看订单表orders。按q键退出。

CREATE TABLE orders (
 order_id INT,
 order_date TIMESTAMP(0),
 customer_name STRING,
 price DECIMAL(10, 5),
 product_id INT,
 order_status BOOLEAN,
 PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
'connector' = 'mysql-cdc',
'hostname' = 'localhost',
'port' = '8527',
'username' = 'polardbx_root',
'password' = '123456',
'database-name' = 'mydb',
'table-name' = 'orders'
);
select * from orders;

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


4、启动压测脚本并实时获取GMV

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

  1. 准备压测脚本。

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

b.执行如下命令,创建配置文件mysql-config.cnf。将如下代码添加到配置文件mysql-config.cnf中。

vim mysql-config.cnf
[client]
user = "polardbx_root"
password = "123456"
host = 127.0.0.1
port = 8527

d.添加完成后的文件内容如下所示。按下Esc键后,输入:wq后按下Enter键保存并退出。

e.执行如下命令,创建脚本buy.sh。将代码添加到脚本buy.sh中。

vim buy.sh
#!/bin/bash
echo "start buying..."
count=0
while :
do
 mysql --defaults-extra-file=./mysql-config.cnf -Dmydb -e "insert into orders values(default, now(), 'free6om', 1024, 102, 0)"
 let count++
 if ! (( count % 10 )); then
  let "batch = count/10"
  echo $batch": got 10 products, gave 1024¥"
 fi
 sleep 0.05
done

g.添加完成后的文件内容如下所示。按下Esc键后,输入:wq后按下Enter键保存并退出。

h.执行如下命令,为脚本buy.sh增加执行权限。

chmod +x buy.sh

2.启动Flink实时计算。

本实验场景通过Flink SQL实时呈现GMV计算结果。

切换至终端一,在Flink中执行如下SQL语句,查询GMV(gmv列)和订单数(orders列)。

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

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

3.启动压测脚本。

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

./buy.sh

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

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





















相关实践学习
快速体验PolarDB开源数据库
本实验环境已内置PostgreSQL数据库以及PolarDB开源数据库:PolarDB PostgreSQL版和PolarDB分布式版,支持一键拉起使用,方便各位开发者学习使用。
目录
相关文章
|
1月前
|
运维 数据处理 Apache
数据实时计算产品对比测评报告:阿里云实时计算Flink版
数据实时计算产品对比测评报告:阿里云实时计算Flink版
|
1月前
|
分布式计算 监控 大数据
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
大数据-148 Apache Kudu 从 Flink 下沉数据到 Kudu
57 1
|
1月前
|
消息中间件 Java Kafka
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
Flink-04 Flink Java 3分钟上手 FlinkKafkaConsumer消费Kafka数据 进行计算SingleOutputStreamOperatorDataStreamSource
46 1
|
1月前
|
SQL 分布式计算 大数据
大数据-108 Flink 快速应用案例 重回Hello WordCount!方案1批数据 方案2流数据(一)
大数据-108 Flink 快速应用案例 重回Hello WordCount!方案1批数据 方案2流数据(一)
46 0
|
1月前
|
大数据 流计算
大数据-108 Flink 快速应用案例 重回Hello WordCount!方案1批数据 方案2流数据(二)
大数据-108 Flink 快速应用案例 重回Hello WordCount!方案1批数据 方案2流数据(二)
45 0
|
2月前
|
SQL 安全 数据处理
揭秘数据脱敏神器:Flink SQL的神秘力量,守护你的数据宝藏!
【9月更文挑战第7天】在大数据时代,数据管理和处理尤为重要,尤其在保障数据安全与隐私方面。本文探讨如何利用Flink SQL实现数据脱敏,为实时数据处理提供有效的隐私保护方案。数据脱敏涉及在处理、存储或传输前对敏感数据进行加密、遮蔽或替换,以遵守数据保护法规(如GDPR)。Flink SQL通过内置函数和表达式支持这一过程。
74 2
|
3月前
|
调度 流计算
Flink 新一代流计算和容错问题之Flink 中的数据可以分为什么类型
Flink 新一代流计算和容错问题之Flink 中的数据可以分为什么类型
|
3月前
|
存储 SQL Cloud Native
揭秘!PolarDB-X存储引擎如何玩转“时间魔术”?Lizard多级闪回技术让你秒回数据“黄金时代”!
【8月更文挑战第25天】PolarDB-X是一款由阿里巴巴自主研发的云原生分布式数据库,以其高性能、高可用性和出色的可扩展性著称。其核心竞争力之一是Lizard存储引擎的多级闪回技术,能够提供高效的数据恢复与问题诊断能力。本文通过一个电商公司的案例展示了一级与二级闪回技术如何帮助快速恢复误删的大量订单数据,确保业务连续性不受影响。一级闪回通过维护最近时间段内历史数据版本链,支持任意时间点查询;而二级闪回则通过扩展数据保留时间并采用成本更低的存储方式,进一步增强了数据保护能力。多级闪回技术的应用显著提高了数据库的可靠性和灵活性,为企业数据安全保驾护航。
44 1
|
3月前
|
SQL 数据库 流计算
Flink CDC数据读取问题之一致性如何解决
Flink CDC 使用Change Data Capture (CDC)技术从数据库捕获变更事件,并利用Flink的流处理能力确保数据读取一致性。相较于传统工具,它具备全增量一体化数据集成能力,满足实时性需求。在实践中解决了高效数据同步、稳定同步大量表数据等问题。应用场景包括实时数据同步、实时数据集成等。快速上手需学习基本概念与实践操作。未来发展方向包括提升效率与稳定性,并依据用户需求持续优化。
125 1

热门文章

最新文章