受欢迎的菏泽网站建设,专业提供网站建设服务,网站建设能赚多少钱,福田公司电话号码多少Flink CDC系列之#xff1a;调研应用Flink CDC将 ELT 从 MySQL 流式传输到 StarRocks方案 准备准备 Flink Standalone 集群准备 docker compose为 MySQL 准备记录使用 Flink CDC CLI 提交作业 同步架构和数据更改路由变更清理 本教程将展示如何使用 Flink CDC 快速构建从 MySQ… Flink CDC系列之调研应用Flink CDC将 ELT 从 MySQL 流式传输到 StarRocks方案 准备准备 Flink Standalone 集群准备 docker compose为 MySQL 准备记录使用 Flink CDC CLI 提交作业 同步架构和数据更改路由变更清理 本教程将展示如何使用 Flink CDC 快速构建从 MySQL 到 StarRocks 的 Streaming ELT 作业包括同步一个数据库的所有表、模式变更演变和将分片表同步到一张表的功能。 本教程中的所有练习都在 Flink CDC CLI 中执行整个过程使用标准 SQL 语法无需一行 Java/Scala 代码或 IDE 安装。
准备
准备一台安装了 Docker 的 Linux 或 MacOS 电脑。
准备 Flink Standalone 集群
下载 Flink 1.18.0 解压得到 flink-1.18.0 目录。
使用以下命令进入 Flink 目录并将 FLINK_HOME 设置为 flink-1.18.0 所在的目录。
cd flink-1.18.0通过将以下参数附加到 conf/flink-conf.yaml 配置文件来启用检查点每 3 秒执行一次检查点。
execution.checkpointing.interval: 3000使用以下命令启动 Flink 集群。
./bin/start-cluster.sh如果启动成功你就可以通过http://localhost:8081/访问Flink Web UI如下所示。 多次执行start-cluster.sh可以启动多个TaskManager。
准备 docker compose
以下教程将使用 docker-compose 准备所需的组件。使用下面提供的内容创建 docker-compose.yml 文件
version: 2.1
services:StarRocks:image: starrocks/allin1-ubuntu:3.2.6ports:- 8080:8080- 9030:9030MySQL:image: debezium/example-mysql:1.1ports:- 3306:3306environment:- MYSQL_ROOT_PASSWORD123456- MYSQL_USERmysqluser- MYSQL_PASSWORDmysqlpwDocker Compose 应包含以下服务容器
MySQL包含一个名为 app_db 的数据库StarRocks存储来自 MySQL 的表
docker-compose up -d该命令会自动以分离模式启动 Docker Compose 配置中定义的所有容器。运行 docker ps 检查这些容器是否正常运行。您也可以访问 http://localhost:8030/ 检查 StarRocks 是否正在运行。
为 MySQL 准备记录
进入 MySQL 容器
docker-compose exec mysql mysql -uroot -p123456创建 app_db 数据库和订单、产品、发货表然后插入记录
-- create database
CREATE DATABASE app_db;USE app_db;-- create orders table
CREATE TABLE orders (
id INT NOT NULL,
price DECIMAL(10,2) NOT NULL,
PRIMARY KEY (id)
);-- insert records
INSERT INTO orders (id, price) VALUES (1, 4.00);
INSERT INTO orders (id, price) VALUES (2, 100.00);-- create shipments table
CREATE TABLE shipments (
id INT NOT NULL,
city VARCHAR(255) NOT NULL,
PRIMARY KEY (id)
);-- insert records
INSERT INTO shipments (id, city) VALUES (1, beijing);
INSERT INTO shipments (id, city) VALUES (2, xian);-- create products table
CREATE TABLE products (
id INT NOT NULL,
product VARCHAR(255) NOT NULL,
PRIMARY KEY (id)
);-- insert records
INSERT INTO products (id, product) VALUES (1, Beer);
INSERT INTO products (id, product) VALUES (2, Cap);
INSERT INTO products (id, product) VALUES (3, Peanut);使用 Flink CDC CLI 提交作业
下载下面列出的二进制压缩包并解压到目录 flink cdc-3.1.0’ flink-cdc-3.1.0-bin.tar.gz flink-cdc-3.1.0 目录下会包含四个目录bin、lib、log、conf。下载下面列出的连接器包并移动到 lib 目录 下载链接只针对稳定版本SNAPSHOT 依赖需要自行基于 master 或 release 分支构建。请注意需要将 jar 移动到 Flink CDC Home 的 lib 目录而不是 Flink Home 的 lib 目录。 MySQL 管道连接器 3.1.0StarRocks 管道连接器 3.1.0
您还需要将 MySQL 连接器放入 Flink lib 文件夹或使用 --jar 参数传递它因为它们不再与 CDC 连接器一起打包
MySQL Connector Java
编写任务配置yaml文件。下面是同步整个数据库的示例文件mysql-to-starrocks.yaml
################################################################################
# Description: Sync MySQL all tables to StarRocks
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*server-id: 5400-5404server-time-zone: UTCsink:type: starrocksname: StarRocks Sinkjdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8080username: rootpassword: table.create.properties.replication_num: 1pipeline:name: Sync MySQL Database to StarRocksparallelism: 2注意
source 中的 tables: app_db..* 通过正则匹配同步 app_db 中的所有表。sink 中的 table.create.properties.replication_num 是因为 Docker 镜像中只有一个 StarRocks BE 节点。
最后使用Cli将作业提交到Flink Standalone集群。
bash bin/flink-cdc.sh mysql-to-starrocks.yaml提交成功后返回信息如下
Pipeline has been submitted to cluster.
Job ID: 02a31c92f0e7bc9a1f4c0051980088a0
Job Description: Sync MySQL Database to StarRocks我们可以通过 Flink Web UI 找到一个名为“Sync MySQL Database to StarRocks“的作业正在运行。 通过Dbeaver等数据库连接工具使用mysql://127.0.0.1:9030连接jdbc可以在StarRocks中查看写入三张表的数据。 同步架构和数据更改
进入MySQL容器
docker-compose exec mysql mysql -uroot -p123456然后修改MySQL中的schema和记录Doris的表也会实时改变
在MySQL中的orders中插入一条记录
INSERT INTO app_db.orders (id, price) VALUES (3, 100.00);在 MySQL 的订单中添加一列
ALTER TABLE app_db.orders ADD amount varchar(100) NULL;从 MySQL 更新订单中的一条记录
UPDATE app_db.orders SET price100.00, amount100.00 WHERE id1;从 MySQL 中删除订单中的一条记录
DELETE FROM app_db.orders WHERE id2;每执行一步刷新一下Dbeaver可以看到StarRocks中展示的订单表会实时更新如下图 同样的通过修改shipping和products表你也可以在StarRocks中实时看到同步修改的结果。
路由变更
Flink CDC 提供了将源表的表结构/数据路由到其他表名的配置。 利用此功能我们可以实现表名、数据库名替换、全库同步等功能。以下是使用路由功能的示例文件
################################################################################
# Description: Sync MySQL all tables to StarRocks
################################################################################
source:type: mysqlhostname: localhostport: 3306username: rootpassword: 123456tables: app_db.\.*server-id: 5400-5404server-time-zone: UTCsink:type: starrocksjdbc-url: jdbc:mysql://127.0.0.1:9030load-url: 127.0.0.1:8030username: rootpassword: table.create.properties.replication_num: 1route:- source-table: app_db.orderssink-table: ods_db.ods_orders- source-table: app_db.shipmentssink-table: ods_db.ods_shipments- source-table: app_db.productssink-table: ods_db.ods_productspipeline:name: Sync MySQL Database to StarRocksparallelism: 2通过上面的路由配置我们可以将app_db.orders的表结构和数据同步到ods_db.ods_orders中从而实现数据库迁移的功能。具体来说source-table支持正则匹配多表来同步分库分表如下
route:- source-table: app_db.order\.*sink-table: ods_db.ods_orders这样我们就可以将app_db.order01、app_db.order02、app_db.order03等分片表同步到一张ods_db.ods_orders表中了。
注意目前还不支持多张表存在相同主键数据的场景后续版本会支持。
清理
完成教程后运行以下命令停止docker-compose.yml目录中的所有容器
docker-compose down在Flink flink-1.18.0目录下执行以下命令停止Flink集群
./bin/stop-cluster.sh
文章转载自： http://www.morning.ckhyj.cn.gov.cn.ckhyj.cn http://www.morning.mfltz.cn.gov.cn.mfltz.cn http://www.morning.znpyw.cn.gov.cn.znpyw.cn http://www.morning.dtfgr.cn.gov.cn.dtfgr.cn http://www.morning.qdlnw.cn.gov.cn.qdlnw.cn http://www.morning.qbkw.cn.gov.cn.qbkw.cn http://www.morning.zrpbf.cn.gov.cn.zrpbf.cn http://www.morning.rjrh.cn.gov.cn.rjrh.cn http://www.morning.hbqfh.cn.gov.cn.hbqfh.cn http://www.morning.tsqpd.cn.gov.cn.tsqpd.cn http://www.morning.wnhsw.cn.gov.cn.wnhsw.cn http://www.morning.ptwqf.cn.gov.cn.ptwqf.cn http://www.morning.rpth.cn.gov.cn.rpth.cn http://www.morning.lwtfr.cn.gov.cn.lwtfr.cn http://www.morning.cljmx.cn.gov.cn.cljmx.cn http://www.morning.qlsyf.cn.gov.cn.qlsyf.cn http://www.morning.nynlf.cn.gov.cn.nynlf.cn http://www.morning.dkgtr.cn.gov.cn.dkgtr.cn http://www.morning.rttkl.cn.gov.cn.rttkl.cn http://www.morning.kdnbf.cn.gov.cn.kdnbf.cn http://www.morning.rlbg.cn.gov.cn.rlbg.cn http://www.morning.qpntn.cn.gov.cn.qpntn.cn http://www.morning.kwblwbl.cn.gov.cn.kwblwbl.cn http://www.morning.rptdz.cn.gov.cn.rptdz.cn http://www.morning.nfzw.cn.gov.cn.nfzw.cn http://www.morning.tpps.cn.gov.cn.tpps.cn http://www.morning.dnpft.cn.gov.cn.dnpft.cn http://www.morning.cqrenli.com.gov.cn.cqrenli.com http://www.morning.nlbw.cn.gov.cn.nlbw.cn http://www.morning.dpppx.cn.gov.cn.dpppx.cn http://www.morning.hwnnh.cn.gov.cn.hwnnh.cn http://www.morning.rqjxc.cn.gov.cn.rqjxc.cn http://www.morning.wchcx.cn.gov.cn.wchcx.cn http://www.morning.pkwwq.cn.gov.cn.pkwwq.cn http://www.morning.cfybl.cn.gov.cn.cfybl.cn http://www.morning.mtbsd.cn.gov.cn.mtbsd.cn http://www.morning.dnhdp.cn.gov.cn.dnhdp.cn http://www.morning.pkmcr.cn.gov.cn.pkmcr.cn http://www.morning.cwlxs.cn.gov.cn.cwlxs.cn http://www.morning.lbrwm.cn.gov.cn.lbrwm.cn http://www.morning.hyxwh.cn.gov.cn.hyxwh.cn http://www.morning.lynmt.cn.gov.cn.lynmt.cn http://www.morning.jjxxm.cn.gov.cn.jjxxm.cn http://www.morning.xbxks.cn.gov.cn.xbxks.cn http://www.morning.3jiax.cn.gov.cn.3jiax.cn http://www.morning.mywmb.cn.gov.cn.mywmb.cn http://www.morning.ylpwc.cn.gov.cn.ylpwc.cn http://www.morning.qykxj.cn.gov.cn.qykxj.cn http://www.morning.xlyt.cn.gov.cn.xlyt.cn http://www.morning.cnprt.cn.gov.cn.cnprt.cn http://www.morning.wfkbk.cn.gov.cn.wfkbk.cn http://www.morning.ykwqz.cn.gov.cn.ykwqz.cn http://www.morning.rgyts.cn.gov.cn.rgyts.cn http://www.morning.zgdnz.cn.gov.cn.zgdnz.cn http://www.morning.msfqt.cn.gov.cn.msfqt.cn http://www.morning.pbpcj.cn.gov.cn.pbpcj.cn http://www.morning.kehejia.com.gov.cn.kehejia.com http://www.morning.plcyq.cn.gov.cn.plcyq.cn http://www.morning.fpkdd.cn.gov.cn.fpkdd.cn http://www.morning.zlsmx.cn.gov.cn.zlsmx.cn http://www.morning.zcwwb.cn.gov.cn.zcwwb.cn http://www.morning.lanyee.com.cn.gov.cn.lanyee.com.cn http://www.morning.gcqs.cn.gov.cn.gcqs.cn http://www.morning.qgwdc.cn.gov.cn.qgwdc.cn http://www.morning.llxns.cn.gov.cn.llxns.cn http://www.morning.kwqt.cn.gov.cn.kwqt.cn http://www.morning.bcngs.cn.gov.cn.bcngs.cn http://www.morning.btqrz.cn.gov.cn.btqrz.cn http://www.morning.mtymb.cn.gov.cn.mtymb.cn http://www.morning.fbqr.cn.gov.cn.fbqr.cn http://www.morning.rtryr.cn.gov.cn.rtryr.cn http://www.morning.snnb.cn.gov.cn.snnb.cn http://www.morning.cpnsh.cn.gov.cn.cpnsh.cn http://www.morning.xzkgp.cn.gov.cn.xzkgp.cn http://www.morning.snnwx.cn.gov.cn.snnwx.cn http://www.morning.ranglue.com.gov.cn.ranglue.com http://www.morning.nmqdk.cn.gov.cn.nmqdk.cn http://www.morning.hbqhz.cn.gov.cn.hbqhz.cn http://www.morning.srrrz.cn.gov.cn.srrrz.cn http://www.morning.lmxzw.cn.gov.cn.lmxzw.cn
