基于 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

创建数据库和表 productsorders,并插入数据

-- 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;

然后, 对于数据库中的表 productsordersshipments, 使用 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.rhkb.cn/news/121507.html

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

相关文章

潜艇来袭(Qt官方案例-2维动画游戏)

一、游戏介绍 1 开始界面 启动程序,进入开始界面。 2 开始新游戏 点击菜单:File》New Game (或者CtrlN)进入新游戏。 开始新游戏之后,会有一个海底的潜艇,和水面舰艇对战。 计算机:自动控制…

STM32f103入门(4)对射式红外传感器计次(外部中断)

中断:在主程序运行过程中,出现了特定的中断触发条件 (中断源),使得CPU暂停当前正在运行的程序,转而去处理中断程序处理完成后又返回原来被暂停的位置继续运行中断优先级:当有多个中断源同时申请中断时,CPU会根据中断源的轻重缓急进…

深度学习推荐系统(四)WideDeep模型及其在Criteo数据集上的应用

深度学习推荐系统(四)Wide&Deep模型及其在Criteo数据集上的应用 在2016年, 随着微软的Deep Crossing, 谷歌的Wide&Deep以及FNN、PNN等一大批优秀的深度学习模型被提出, 推荐系统全面进入了深度学习时代, 时至今日&#x…

聚焦磷酸铁锂产线革新,宏工科技一站式解决方案

兼顾了低成本与安全性两大属性,磷酸铁锂市场在全球范围内持续升温,并有望保持较高的景气度。巨大的需求空间之下,行业对于锂电装备企业的自动化与智能化水平、整线交付能力、产品效率与稳定性等均提出了新的要求。 以宏工科技股份有限公司&a…

C#通过ModbusTcp协议读写西门子PLC中的浮点数

一、Modbus TCP通信概述 MODBUS/TCP是简单的、中立厂商的用于管理和控制自动化设备的MODBUS系列通讯协议的派生产品,显而易见,它覆盖了使用TCP/IP协议的“Intranet”和“Internet”环境中MODBUS报文的用途。协议的最通用用途是为诸如PLC,I/…

LiveGBS流媒体平台GB/T28181功能-支持数据库切换为高斯数据库信创瀚高数据信创数据库

LiveGBS流媒体平台GB/T28181功能-支持数据库切换为高斯数据库信创瀚高数据信创数据库 1、如何配置切换高斯数据库?2、如何配置切换信创瀚高数据库?3、搭建GB28181视频直播平台 1、如何配置切换高斯数据库? livecms.ini -> [db]下面添加配…

华为Mate60低调发布,你所不知道的高调真相?

华为Mate60 pro 这两天的劲爆新闻想必各位早已知晓,那就是华为Mate60真的来了!!!并且此款手机搭载了最新国产麒麟9000s芯片,该芯片重新定义了手机性能的巅峰。不仅在Geekbench测试中表现出色,还在实际应用…

成都瀚网科技有限公司:抖店的评论会消失吗?

抖店是抖音推出的电子商务平台。很多用户在购物后都会对产品进行评价。但有时用户可能会发现抖店评论缺失,让用户产生一些疑惑和困惑。本文将围绕这个问题提供一些答案和解决方案。 1.为什么抖店评论不见了? 首先需要明确的是,抖店评论消失可…

单值二叉树

目录 题目题目要求示例 解答方法一、实现思路时间复杂度和空间复杂度代码 方法二、实现思路时间复杂度和空间复杂度代码 题目 单值二叉树 题目要求 题目链接 示例 解答 方法一、 递归 实现思路 时间复杂度和空间复杂度 时间复杂度:O(N) 空间复杂度&#xf…

Jenkins详解(三)

Jenkins详解(三) 目录 Jenkins详解(三) 1、Jenkins介绍2、Jenkins CI/CD 流程3、部署环境 3.1 环境准备3.2 安装GitLab3.3 初始化GitLab3.4 GitLab中文社区版补丁包安装3.5 修改GitLab配置文件/etc/gitlab/gitlab.rb3.6 在宿主机输入 http://192.168.200.26:88 地址就可以访问了…

什么是AJAX?如何使用原生JavaScript搭建AJAX请求?

聚沙成塔每天进步一点点 ⭐ 专栏简介⭐ AJAX(Asynchronous JavaScript and XML)⭐ 原生JavaScript中的AJAX请求1. 创建XMLHttpRequest对象2. 配置请求3. 设置回调函数4. 发送请求 ⭐ 完整示例⭐ 写在最后 ⭐ 专栏简介 前端入门之旅:探索Web开…

实现Android APK瘦身99.99%

摘要: 如何瘦身是 APK 的重要优化技术。APK 在安装和更新时都需要经过网络下载到设备,APK 越小,用户体验越好。本文作者通过对 APK 内在机制的详细解析,给出了对 APK 各组成成分的优化方法及技术,并实现了一个基本 APK…

Unity 安卓(Android)端AVProVideo插件播放不了视频,屏幕一闪一闪的

编辑器运行没有问题,但是安卓就有问题,在平板上运行就会报错: vulkan graphics API is notsupported 说不支持Vulkan图形API,解决方法:把Vulkan删除掉

QTableView合并单元格

QtableView的功能 QTableView是Qt框架提供的用于显示表格数据的类。它是基于MVC(模型-视图-控制器)设计模式的一部分,用于将数据模型和界面视图分离。 以下是一些QTableView的主要特点和功能: 1. 显示表格数据: QTa…

手写Mybatis:第20章-Mybatis 框架源码10种设计模式分析

文章目录 一、类型:创建型模式1.1 工厂模式1.2 单例模式1.3 建造者模式 二、类型:结构型模式2.1 适配器模式2.2 代理模式2.3 组合模式2.4 装饰器模式 三、类型:行为型模式3.1 模板模式3.2 策略模式3.3 迭代器模式 一、类型:创建型…

题①拷贝构造相关笔试题

问:此代码中有几次构造,几次拷贝构造? W f(W u) {W v(u);W w v;return w; } int main() {w x;w y f(x);return 0;解析:一次构造,四次拷贝构造。 再来一题 W f(W u)…

Redis未授权访问漏洞复现

Redis 简单使用 Redis 未设置密码,客户端工具可以直接链接。 Redis 是非关系型数据库系统,没有库表列的逻辑结构,仅仅以键值对的方式存储数据。 先启动容器 Redis 未设置密码,客户端工具可以直接链接 https://github.com/xk11z/…

Laravel 模型的关联查询 Debugbar 调试器 模型的预加载 ⑩②

作者 : SYFStrive 博客首页 : HomePage 📜: THINK PHP 📌:个人社区(欢迎大佬们加入) 👉:社区链接🔗 📌:觉得文章不错可以点点关注 &#x1f44…

冠达管理:券商8月调研热情高 工业机械行业受青睐

截至9月4日记者发稿,8月以来券商累计调研次数约1.44万次,环比增加超160%。其间,工业机械职业公司获券商调研最多。 调研逾900只个股 截至9月4日发稿,8月以来券商累计调研948只个股。从个股调研热度看,容百科技最受券…

【ES6】JavaScript中的Symbol

Symbol是JavaScript中的一种特殊的、不可变的、不可枚举的数据类型。它通常用于表示一个唯一的标识符,可以作为对象的属性键,确保对象的属性键的唯一性和不可变性。 Symbol.for()是Symbol的一个方法,它用于创建一个已经注册的Symbol对象。当…