基于 Flink CDC 构建 MySQL 和 Postgres 的 Streaming ETL

官方网址：https://ververica.github.io/flink-cdc-connectors/release-2.3/content/%E5%BF%AB%E9%80%9F%E4%B8%8A%E6%89%8B/mysql-postgres-tutorial-zh.html官方教程有些坑，经过自己实测，记录个笔记。

服务器环境：

VM虚拟机：CentOS7.9

docker版本：Docker version 24.0.5, build ced0996

docker compose 版本：2.19

jdk 1.8

虚拟机IP:192.168.122.131 

内存：16G(一定要大于等于16G)

CPU：4g

磁盘 ：>= 60G

一、docker  compose安装

DOCKER_CONFIG=${DOCKER_CONFIG:-/usr/local/lib/docker/cli-plugins}
mkdir -p $DOCKER_CONFIG/cli-plugins
curl -SL https://github.com/docker/compose/releases/download/v2.19.1/docker-compose-linux-x86_64 -o $DOCKER_CONFIG/cli-plugins/docker-compose

对文件应用可执行权限：

chmod +x $DOCKER_CONFIG/cli-plugins/docker-compose

测试安装是否成功

docker compose version #之前的v1版本命令是docker-compose --version

参考：https://blog.csdn.net/qq_40099908/article/details/131611496

二、实战

这篇教程将展示如何基于 Flink CDC 快速构建 MySQL 和 Postgres 的流式 ETL。本教程的演示都将在 Flink SQL CLI 中进行，只涉及 SQL，无需一行 Java/Scala 代码，也无需安装 IDE。

假设我们正在经营电子商务业务，商品和订单的数据存储在 MySQL 中，订单对应的物流信息存储在 Postgres 中。 对于订单表，为了方便进行分析，我们希望让它关联上其对应的商品和物流信息，构成一张宽表，并且实时把它写到 ElasticSearch 中。

接下来的内容将介绍如何使用 Flink Mysql/Postgres CDC 来实现这个需求，系统的整体架构如下图所示：

1、准备教程所需要的组件

接下来的教程将以 docker-compose 的方式准备所需要的组件。

使用下面的内容创建一个 docker-compose.yml 文件：

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

该 Docker Compose 中包含的容器有：

  • MySQL: 商品表 products 和 订单表 orders 将存储在该数据库中， 这两张表将和 Postgres 数据库中的物流表 shipments进行关联，得到一张包含更多信息的订单表 enriched_orders

  • Postgres: 物流表 shipments 将存储在该数据库中

  • Elasticsearch: 最终的订单表 enriched_orders 将写到 Elasticsearch

  • Kibana: 用来可视化 ElasticSearch 的数据

在 docker-compose.yml 所在目录下执行下面的命令来启动本教程需要的组件：

docker compose up -d

该命令将以 detached 模式自动启动 Docker Compose 配置中定义的所有容器。你可以通过 docker ps 来观察上述的容器是否正常启动了，也可以通过访问 http://192.168.122.131:5601来查看 Kibana 是否运行正常。

2、下载 Flink 和所需要的依赖包

下载 Flink 1.16.0 并将其解压至目录 flink-1.16.0  ，

下载下面列出的依赖包，并将它们放到目录 flink-1.16.0/lib/ 下：

  1. 下载链接只对已发布的版本有效, SNAPSHOT 版本需要本地编译

    • flink-sql-connector-elasticsearch7-1.16.0.jar

    • flink-sql-connector-mysql-cdc-2.3.0.jar

    • flink-sql-connector-postgres-cdc-2.3.0.jar

准备数据

在 MySQL 数据库中准备数据

进入 MySQL 容器

docker compose exec mysql mysql -uroot -p123456

创建数据库和表 products，orders，并插入数据

-- MySQL
CREATE DATABASE mydb;
USE mydb;
CREATE TABLE products (id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,name VARCHAR(255) NOT NULL,description VARCHAR(512)
);
ALTER TABLE products AUTO_INCREMENT = 101;INSERT INTO products
VALUES (default,"scooter","Small 2-wheel scooter"),(default,"car battery","12V car battery"),(default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),(default,"hammer","12oz carpenter's hammer"),(default,"hammer","14oz carpenter's hammer"),(default,"hammer","16oz carpenter's hammer"),(default,"rocks","box of assorted rocks"),(default,"jacket","water resistent black wind breaker"),(default,"spare tire","24 inch spare tire");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,order_status BOOLEAN NOT NULL -- Whether order has been placed
) 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);

注意：mysql会遇到时区不对的情况。

在mysql容器调整时区：

set time_zone='+8:00';
SET GLOBAL time_zone = '+8:00';
flush privileges;
SELECT @@global.time_zone;
show variables like '%time_zone%';

在 Postgres 数据库中准备数据

进入 Postgres 容器

docker compose exec postgres psql -h localhost -U postgres

创建表 shipments，并插入数据

-- PG
CREATE TABLE shipments (shipment_id SERIAL NOT NULL PRIMARY KEY,order_id SERIAL NOT NULL,origin VARCHAR(255) NOT NULL,destination VARCHAR(255) NOT NULL,is_arrived BOOLEAN NOT NULL
);
ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001;
ALTER TABLE public.shipments REPLICA IDENTITY FULL;
INSERT INTO shipments
VALUES (default,10001,'Beijing','Shanghai',false),(default,10002,'Hangzhou','Shanghai',false),(default,10003,'Shanghai','Hangzhou',false);

启动 Flink 集群和 Flink SQL CLI

使用下面的命令跳转至 Flink 目录下

cd flink-1.16.0

使用下面的命令启动 Flink 集群

./bin/start-cluster.sh

启动成功的话，可以在 http://192.168.122.131:8081/ 访问到 Flink Web UI，如下所示：

注：若在VM之外的本地的电脑里无法访问，则需要调整 /flink-1.16.0/conf/flink-conf.yaml文件，

将rest.address值改为：0.0.0.0

开放单个端口（开放后需要要重启防火墙才生效） ；

firewall-cmd --zone=public --add-port=8081/tcp --permanent

重启防火墙 ； systemctl restart firewalld

  另：还有个参数taskmanager.numberOfTaskSlots: 50，一般设置大一些的值，比如50。

使用下面的命令启动 Flink SQL CLI

./bin/sql-client.sh

启动成功后，可以看到如下的页面：

在 Flink SQL CLI 中使用 Flink DDL 创建表

首先，开启 checkpoint，每隔3秒做一次 checkpoint

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

然后, 对于数据库中的表 products, orders, shipments， 使用 Flink SQL CLI 创建对应的表，用于同步这些底层数据库表的数据

-- Flink SQL
Flink SQL> CREATE TABLE products (id INT,name STRING,description STRING,PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'mysql-cdc','hostname' = 'localhost','port' = '3306','username' = 'root','password' = '123456','database-name' = 'mydb','table-name' = 'products');Flink SQL> 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' = '3306','username' = 'root','password' = '123456','database-name' = 'mydb','table-name' = 'orders');Flink SQL> CREATE TABLE shipments (shipment_id INT,order_id INT,origin STRING,destination STRING,is_arrived BOOLEAN,PRIMARY KEY (shipment_id) NOT ENFORCED) WITH ('connector' = 'postgres-cdc','hostname' = 'localhost','port' = '5432','username' = 'postgres','password' = 'postgres','database-name' = 'postgres','schema-name' = 'public','table-name' = 'shipments');

最后，创建 enriched_orders 表， 用来将关联后的订单数据写入 Elasticsearch 中

-- Flink SQL
Flink SQL> CREATE TABLE enriched_orders (order_id INT,order_date TIMESTAMP(0),customer_name STRING,price DECIMAL(10, 5),product_id INT,order_status BOOLEAN,product_name STRING,product_description STRING,shipment_id INT,origin STRING,destination STRING,is_arrived BOOLEAN,PRIMARY KEY (order_id) NOT ENFORCED) WITH ('connector' = 'elasticsearch-7','hosts' = 'http://localhost:9200','index' = 'enriched_orders');

关联订单数据并且将其写入 Elasticsearch 中

使用 Flink SQL 将订单表 order 与 商品表 products，物流信息表 shipments 关联，并将关联后的订单信息写入 Elasticsearch 中

-- Flink SQL
Flink SQL> INSERT INTO enriched_ordersSELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrivedFROM orders AS oLEFT JOIN products AS p ON o.product_id = p.idLEFT JOIN shipments AS s ON o.order_id = s.order_id;

现在，就可以在 Kibana 中看到包含商品和物流信息的订单数据。

首先访问 http://192.168.122.131:5601/app/kibana#/management/kibana/index_pattern 创建 index pattern enriched_orders.

然后就可以在 http://192.168.122.131:5601/app/kibana#/discover 看到写入的数据了.

接下来，修改 MySQL 和 Postgres 数据库中表的数据，Kibana中显示的订单数据也将实时更新：

在 MySQL 的 orders 表中插入一条数据

--MySQL
INSERT INTO orders
VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);

在 Postgres 的 shipment 表中插入一条数据

--PG
INSERT INTO shipments
VALUES (default,10004,'Shanghai','Beijing',false);

在 MySQL 的 orders 表中更新订单的状态

--MySQL
UPDATE orders SET order_status = true WHERE order_id = 10004;

在 Postgres 的 shipment 表中更新物流的状态

--PG
UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;

在 MYSQL 的 orders 表中删除一条数据

--MySQL
DELETE FROM orders WHERE order_id = 10004;

每执行一步就刷新一次 Kibana，可以看到 Kibana 中显示的订单数据将实时更新，如下所示：

环境清理

本教程结束后，在 docker-compose.yml 文件所在的目录下执行如下命令停止所有容器：

docker compose down

在 Flink 所在目录 flink-1.16.0 下执行如下命令停止 Flink 集群：

./bin/stop-cluster.sh

异常排查

若数据异常，在flink的网页里看查看错误信息。

http://192.168.122.131:8081/#/job/running

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.hqwc.cn/news/102735.html

如若内容造成侵权/违法违规/事实不符，请联系编程知识网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

护航数字政府建设，美创科技成为“数字政府建设赋能计划”成员单位

护航数字政府建设，美创科技成为“数字政府建设赋能计划”成员单位

近日，“2023软博会-软件驱动数字政府创新发展论坛”顺利召开，本次论坛由中国信息通信研究院、中国通信标准化协会承办，中国通信标准化协会云计算标准和开源推进委员会、数字政府建设赋能计划支持。 天津市工业和信息化局总经济师杨冬梅、中国…
阅读更多...
Paimon+StarRocks 湖仓一体数据分析方案

Paimon+StarRocks 湖仓一体数据分析方案

本文整理自阿里云高级开发工程师曾庆栋（曦乐）在 Streaming Lakehouse Meetup 分享的内容，深入探讨了传统数据仓库分析、PaimonStarRocks湖仓一体数据分析、StarRocks 与 Paimon 的协同使用方法与实现原理，以及StarRocks 社区湖仓分…
阅读更多...
无涯教程-JavaScript - DELTA函数

无涯教程-JavaScript - DELTA函数

描述 DELTA函数测试两个值是否相等。如果number1 number2,则返回1；否则返回1。否则返回0。 您可以使用此功能来过滤一组值。如,通过合计几个DELTA函数,您可以计算相等对的计数。此功能也称为Kronecker Delta功能。 语法 DELTA (number1, [number2])争论 Argum…
阅读更多...
python调用GPT实现：智能用例生成工具

python调用GPT实现：智能用例生成工具

工具作用： 根据输入的功能点，生成通用测试点 实现步骤 工具实现主要分2个步骤： 1.https请求调用Gpt,将返回响应结果保存为.md文件 2.用python实现 将 .md文件转换成.xmind文件 3.写个简单的前端页面，调用上述步骤接口 详细代…
阅读更多...
table 单元格中嵌套子表格 样式撑开问题

table 单元格中嵌套子表格 样式撑开问题

如图，表格中的td嵌套表格，里边表格把外层撑开，不能按100%显示； 解决办法 给父级table 加一个table-layout:fixed;样式
阅读更多...
(其他) 剑指 Offer 46. 把数字翻译成字符串 ——【Leetcode每日一题】

(其他) 剑指 Offer 46. 把数字翻译成字符串 ——【Leetcode每日一题】

❓ 剑指 Offer 46. 把数字翻译成字符串 难度：中等 给定一个数字，我们按照如下规则把它翻译为字符串：0 翻译成 “a” ，1 翻译成 “b”，……，11 翻译成 “l”，……，25 翻译成 “z”。…
阅读更多...
cpolar内网穿透

cpolar内网穿透

目录 一、引言二、什么是cpolar三、内网穿透四、如何使用cpolar1、下载cpolar软件安装包2、注册cpolar账号3、使用cpolar 一、引言 当我们完成了一个tomcat的web项目之后，如果我们想让其他电脑访问到这个项目，我们可以让其他电脑和本机连接到同一个局域…
阅读更多...
leetcode刷题--栈与递归

leetcode刷题--栈与递归

文章目录 1. 682 棒球比赛2. 71 简化路径3. 388 文件的最长绝对路径4. 150 逆波兰表达式求值5. 227. 基本计算器II6. 224. 基本计算器7. 20. 有效的括号8. 636. 函数的独占时间9. 591. 标签验证器10. 32.最长有效括号12. 341. 扁平化嵌套列表迭代器13. 394.字符串解码 1. 682 棒…
阅读更多...
MySQL 如何避免 RC 隔离级别下的 INSERT 死锁？

MySQL 如何避免 RC 隔离级别下的 INSERT 死锁？

本文分析了 INSERT 及其变种（REPLACE/INSERT ON DUPLICATE KEY UPDATE）的几个场景的死锁及如何避免。 作者：张洛丹，DBA 数据库技术爱好者~ 爱可生开源社区出品，原创内容未经授权不得随意使用，转载请联系小编…
阅读更多...
win11和虚拟机上的ubuntu系统共享文件夹

win11和虚拟机上的ubuntu系统共享文件夹

出发点：有时候需要在虚拟机内和win11系统之间进行文件拷贝操作，但是虚拟机内的Vmware Tools不好用 解决方式 开启虚拟机共享文件夹 在虚拟机的Ubuntu系统内的终端命令行输入以下命令 sudo /usr/bin/vmhgfs-fuse .host:/SVMware /mnt/win -o subtype…
阅读更多...
电子产品CE认证申请，CE-RED认证

电子产品CE认证申请，CE-RED认证

CE认证，即只限于产品不危及人类、动物和货品的安全方面的基本安全要求，而不是一般质量要求，协调指令只规定主要要求，一般指令要求是标准的任务。 构成欧洲指令核心的"主要要求"，在欧共体1985年5月7日的&…
阅读更多...
【webpack】HMR热更新原理

【webpack】HMR热更新原理

本文：参考文章 一、HMR是什么，为什么出现 1、出现的原因 之前，应用的加载、更新都是一个页面级别的操作，即使单个代码文件更新，整个页面都要刷新，才能拿到最新的代码同步到浏览器，导致会丢失…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023