通过 Flink CDC CLI 提交任务：整库同步 MySQL 到 Doris

最新推荐文章于 2025-03-13 16:41:33 发布
最新推荐文章于 2025-03-13 16:41:33 发布
阅读量1.2k 收藏 25
点赞数 20
文章标签： flink mysql 大数据
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/qq_42978535/article/details/145129405
版权

Apache Flink 是一款流处理框架，Flink CDC（Change Data Capture）是 Flink 提供的一个开源工具，它可以实时捕获数据库（如 MySQL）中的数据变更并将这些变更流式地传输到其他系统中（如 Doris）。通过 Flink CDC，可以实现从 MySQL 到 Doris 的数据同步。

本文将详细讲解如何通过 Flink CDC CLI 提交任务，实现整库同步 MySQL 到 Doris。我们将展示详细的配置和代码。

前提条件

  • Flink 集群：确保 Flink 集群已经搭建完成。
  • Doris 集群：确保 Doris 集群已经搭建完成。
  • MySQL 数据库：需要从 MySQL 中同步数据。
  • Flink CDC jar 包：确保下载并配置了 Flink CDC 的相关 jar 包。

步骤概览

  1. 配置 Flink CDC 连接器。
  2. 编写 Flink CDC 作业代码。
  3. 提交 Flink CDC 任务到 Flink 集群。
  4. 监控任务执行。

1. 配置 Flink CDC 连接器

首先，需要在 Flink 配置文件中添加 Flink CDC 相关的连接器。Flink CDC 提供了 MySQL 和 Doris 的连接器，确保将相关的连接器 jar 包放到 Flink 集群的 lib 目录中。

下载 Flink CDC 依赖

你可以从 Flink CDC 官方 GitHub 下载适合的版本。

在项目中使用 Maven 来集成依赖：

<dependency>
    <groupId>com.ververica</groupId>
    <artifactId>flink-connector-mysql-cdc</artifactId>
    <version>2.0.0</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-connector-doris</artifactId>
    <version>1.13.0</version>
</dependency>

2. 编写 Flink CDC 作业代码

通过 Flink CDC，你可以实现 MySQL 数据库到 Doris 的整库同步。Flink CDC 会捕获 MySQL 数据库的增量变化（如 INSERT、UPDATE、DELETE 操作），并将这些变化实时地推送到 Doris。

以下是一个简单的 Flink CDC 作业，展示如何将 MySQL 数据库的变更同步到 Doris：

MySQL 到 Doris 的同步代码

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.doris.DorisSink;
import org.apache.flink.streaming.connectors.mysql.MySQLSource;
import org.apache.flink.streaming.connectors.mysql.table.MySQLTableSource;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcStatementBuilder;
import org.apache.flink.connector.jdbc.JdbcSink;

public class MySQLToDorisSync {

    public static void main(String[] args) throws Exception {
        // 创建 Flink 流式环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // MySQL 配置
        String mysqlHost = "localhost";
        String mysqlPort = "3306";
        String mysqlDatabase = "your_database";
        String mysqlUser = "your_username";
        String mysqlPassword = "your_password";
        
        // Doris 配置
        String dorisHost = "localhost";
        String dorisPort = "8030";  // Doris FE 服务端口
        String dorisDatabase = "your_doris_database";
        String dorisTable = "your_doris_table";
        
        // 从 MySQL 获取增量数据
        MySQLSource<String> mySQLSource = MySQLSource.<String>builder()
                .hostname(mysqlHost)
                .port(Integer.parseInt(mysqlPort))
                .databaseList(mysqlDatabase)
                .tableList("your_database.*") // 选择需要同步的表
                .username(mysqlUser)
                .password(mysqlPassword)
                .startupMode(MySQLSource.StartupMode.LATEST_OFFSET) // 从最新的偏移量开始同步
                .deserializer(new MySQLRowDataDebeziumDeserializationSchema())  // 自定义反序列化 schema
                .build();

        // 设置 Flink 任务的数据源
        DataStream<String> mysqlStream = env.fromSource(mySQLSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source");

        // 将数据流写入到 Doris
        DorisSink<String> dorisSink = DorisSink.<String>builder()
                .setHost(dorisHost)
                .setPort(dorisPort)
                .setDatabase(dorisDatabase)
                .setTable(dorisTable)
                .setUsername("doris_user")
                .setPassword("doris_password")
                .build();

        // 向 Doris 写数据
        mysqlStream.sinkTo(dorisSink);

        // 执行 Flink 作业
        env.execute("MySQL to Doris Sync");
    }
}

关键点解释：

  • MySQLSource：从 MySQL 中读取数据，通过 Flink CDC 捕获 MySQL 中的数据变更。
  • DorisSink：将捕获到的数据变更写入到 Doris 集群。
  • StartupMode.LATEST_OFFSET：表示从 MySQL 数据库的最新位置开始同步。
  • 自定义反序列化 Schema：用于将 MySQL 的变更事件转换为 Flink 可识别的数据格式（此处使用了 MySQLRowDataDebeziumDeserializationSchema）。

3. 提交 Flink CDC 任务到 Flink 集群

完成代码编写后，我们可以通过 Flink CLI 提交任务到 Flink 集群。

3.1 编译和打包

首先，使用 Maven 打包 Flink 作业：

mvn clean package -DskipTests

3.2 提交作业

使用 Flink CLI 提交作业：

flink run -c com.example.MySQLToDorisSync /path/to/your/flink-job.jar

3.3 监控任务

Flink 提供了 Web UI 用于监控任务执行情况。通过浏览器访问 Flink Web UI（通常是 http://localhost:8081），可以查看作业的状态、任务的执行情况、以及各个节点的资源利用情况。

4. 完整代码示例

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.doris.DorisSink;
import org.apache.flink.streaming.connectors.mysql.MySQLSource;
import org.apache.flink.streaming.connectors.mysql.table.MySQLTableSource;
import org.apache.flink.connector.jdbc.JdbcConnectionOptions;
import org.apache.flink.connector.jdbc.JdbcSink;
import org.apache.flink.streaming.api.datastream.DataStream;

public class MySQLToDorisSync {

    public static void main(String[] args) throws Exception {
        // Flink 环境初始化
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // MySQL 配置
        String mysqlHost = "localhost";
        String mysqlPort = "3306";
        String mysqlDatabase = "your_database";
        String mysqlUser = "your_username";
        String mysqlPassword = "your_password";
        
        // Doris 配置
        String dorisHost = "localhost";
        String dorisPort = "8030";
        String dorisDatabase = "your_doris_database";
        String dorisTable = "your_doris_table";

        // MySQL CDC 源
        MySQLSource<String> mysqlSource = MySQLSource.<String>builder()
                .hostname(mysqlHost)
                .port(Integer.parseInt(mysqlPort))
                .databaseList(mysqlDatabase)
                .tableList("your_database.*")
                .username(mysqlUser)
                .password(mysqlPassword)
                .startupMode(MySQLSource.StartupMode.LATEST_OFFSET)
                .deserializer(new MySQLRowDataDebeziumDeserializationSchema())
                .build();

        // 创建数据流
        DataStream<String> mysqlStream = env.fromSource(mysqlSource, WatermarkStrategy.noWatermarks(), "MySQL CDC Source");

        // Doris Sink 配置
        DorisSink<String> dorisSink = DorisSink.<String>builder()
                .setHost(dorisHost)
                .setPort(dorisPort)
                .setDatabase(dorisDatabase)
                .setTable(dorisTable)
                .setUsername("doris_user")
                .setPassword("doris_password")
                .build();

        // 将数据写入 Doris
        mysqlStream.sinkTo(dorisSink);

        // 执行 Flink 作业
        env.execute("MySQL to Doris Sync");
    }
}

总结

通过 Flink CDC，我们可以实现 MySQL 到 Doris 的整库同步。这个流程的关键是配置好 MySQL 的增量捕获、并将数据变更流式地写入 Doris。通过 Flink CLI，我们可以方便地提交作业并监控任务执行情况。这个技术方案可以在数据同步、实时数据仓库等场景中应用。

Flink CDC 同步数据到Doris
03-29 7万+
实时同步支持批量同步支持路由，和一些自定义选项支持schema 变更支持自动建表。
Flink CDC同步mysql数据到doris
qq_36070104的博客
10-13 1648
Flink CDC 是一个基于流的数据集成工具，旨在为用户提供一套功能更加全面的编程接口（API）。该工具使得用户能够以 YAML 配置文件的形式，优雅地定义其 ETL（Extract, Transform, Load）流程，并协助用户自动化生成定制化的 Flink 算子并且提交 Flink 作业。
使用 Flink CDC 实现 MySQL 数据,表结构实时入 Apache Doris
希夷的博客
07-12 5596
实现mysql到doris的增量同步，新加入表全量同步，实时修改表结构
Flink1.15.4+flinkcdc+Dinky0.7.3+Mysql5.7+Doris1.2.1实现实时同步
qq_37247664的博客
08-09 935
需要将dinky的dlink-common-0.7.3.jar,dlink-client-1.15-0.7.3.jar,dlink-client-base-0.7.3.jar 这三个jar包放入flink的lib下。修改conf下的flink的yaml文件中rest.bind-address:0.0.0.0。'sink.label-prefix' = '唯一'访问：ip:8888。'database-name' = '库名','table-name' = '表名''username' = '账号',
基于Flink CDC实现Mysql实时同步到Doris系列教程一：Flink环境的部署
遇码开源社区
03-12 494
前面为大家介绍了Flink CDC，我们已经知道Flink CDC 是一个基于流的数据集成工具。本文开始，将会手把手的带领大家动手完成把Mysql的数据实时同步到Doris中，有兴趣的同学可以持续关注本系列教程。
Flink CDC实时同步MySQL到Doris
weixin_42828342的博客
09-28 3080
Apache Flink CDC（Change Data Capture）是一个用于捕获和跟踪数据库更改的技术，它能够实时地从数据库中获取数据变更，并将这些变更流式传输到其他系统进行处理和分析。Flink CDC 提供了对多种数据库的支持，包括 MySQL、Doris、MongoDB 等，Flink CDC 在大数据中主要用于实时数据同步、实时数据分析、实时监控和告警、日志和审计记录、流式ETL、实时事件处理等方面。
FlinkCDC 3.1.0 与 Flink 1.18.0 安装及使用 Mysql To Doris 整库同步，使用 pipepline连接器
SunTecTec
06-20 1979
bin/flink-cdc.sh 会用到 linux的系统环境变量（vim /etc/profile配置），使用环境变量 FLINK_HOME。问题：flink lib 加入 mysql-connect-java.jar 否则flink任务会无法连接sink端，报错Exception。解决：flink lib 安装完成后，引入 mysql-connect-java.jar ，配置如下。4、查看 flinkcdc log：同步任务提交成功。
Flink CDC案例：MySQL到Doris实时整库同步（提交flink job jar方式）
学亮编程手记
12-17 2572
本示例是借助插件实现通过 Flink 同步mysql数据（读取、插入、修改、删除）到Doris 中存储。最佳实践：使用 Flink Doris Connector 最适合的场景就是实时/批次同步源数据（Mysql，Oracle，PostgreSQL 等）到 Doris，使用 Flink 对 Doris 中的数据和其他数据源进行联合分析，这里介绍Mysql到doris的数据接入方式。
使用 Flink CDC 实现 MySQL 数据实时入 Apache Doris
修破立生
01-07 1913
使用 Flink CDC 实现 MySQL 数据实时入 Apache Doris, Apache Doris使用唯一模型，实现更新/删除操作
Flink CDC2.4 整库实时同步MySql 到Doris
希夷的博客
06-29 6694
mysql 到doris的数据、表结构实时同步
基于Flink CDC实现Mysql实时同步到Doris系列教程五：Flink CDC提交任务并验证
最新发布
遇码开源社区
03-13 1070
本文也是该系列教程的最后一篇，期望本系列教程可以实实在在的帮助到你。本文将介绍如何把YAML任务文件使用提交到Flink环境中，最后我们再验证doris数据库中的数据，查看是否实时可查询。本系列教程到这里就结束了，用非常有限的图文带领大家体验了Flink CDC的能力。但是对于我们个人的学习而言则才刚刚开始。但是也希望大家可以理解，大家的环境不同、操作不同，我也没有办法一一帮助大家解决所有的问题，期望大家遇到问题不要着急，慢慢来，解决问题的过程也是深度学习的过程。最后再次强调——多动手。
Flink CDC实践
wwplcy的博客
02-27 1675
访问http://虚拟机ip:8081，查看flink webUI是否正常访问。下载flink安装包，网速较慢，可以尝试在国内镜像网址。将mysql和doris的驱动包移动到lib目录下。启动flink，查看启动是否成功。解压flink-cdc安装包。配置flink环境变量。下载flink安装包。解压flink安装包。
Flinkcdc（一）同步mysql的binlog到doris
jwandbj的博客
12-25 993
现在使用第一种方式来实现，通过在flink的client里面进行sql方式的执行。在flink的bin目录下有一个./bin/sql-client.sh，进入到这里面可以执行sql语句。flink利用mysql-cdc从mysql读取binlog日志；一种使用的是datastream的方式；一种是直接使用sql的方式；
flink-cdc同步mysql到doris建设数据仓储最佳实践
AlaiaCodeingLife
12-04 1070
当前需要搭建数据仓储，横向对比了多个数据库最终选定doris，正常的单表，多表数据同步已经完成开发。但是单全量表结构同步还是没有完成，发现flink有这个功能，现在进行使用。
FlinkCDC如何实时同步mysql数据到DorisDB/StarRocks
Demon的博客
09-19 3910
FlinkCDC同步mysql数据到DorisDB/StarRocks
FlinkCDC（二）同步mysql的binlog到doris
jwandbj的博客
12-25 1132
前面一章使用flink sql的方式来做数据的同步，但是有一个问题，如果我们一个库里面有100多张表或者更多，那就要写100多个create table来连接，mysql的连接数也会一下子暴涨，所以对于多个表或者整个库或者多库的数据同步，只能使用datastream来同步。这个项目部需要用到spirng、springboot和springclound，直接在main方法里面实现逻辑；先创建一个project，取名flinkcdc-mysql-doris。第一步先创建flink datastream的环境。
FlinkCDC读取mysql的表写入kafka/doris/iceberg
qq_38151907的博客
04-04 1188
FlinkCDC读取mysql的表写入kafka/doris/iceberg
评论
添加红包

请填写红包祝福语或标题

个

红包个数最小为10个

元

红包金额最低5元

当前余额3.43元 前往充值 >
需支付：10.00元
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包

打赏作者

一只蜗牛儿

你的鼓励将是我创作的最大动力

¥1 ¥2 ¥4 ¥6 ¥10 ¥20
扫码支付：¥1
获取中
扫码支付

您的余额不足，请更换扫码支付或充值

打赏作者

实付元
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明：

1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载，可以购买VIP、付费专栏及课程。

余额充值