暂无图片
暂无图片
暂无图片
暂无图片
暂无图片

【Apache Doris】如何一键实现MySQL万表整库同步?

一臻数据 2023-12-03
68

  导读   本文主要分享 Apache Doris 是如何一键实现MySQL万表整库同步的。

全文目录:

  1. 环境信息

  2. 功能介绍

  3. 前提概要

  4. 快速体验

  5. 常见问题



环境信息

1. 硬件信息

  • CPU:48C

  • 内存:256G

2. 软件信息

  • 系统:CentOS

  • Apache Doris版本:2.0.2

  • MySQL版本:5.7

  • Flink版本:1.17.1

  • Flink-Doris-Connector版本:1.5

功能介绍

Mysql万表存量+增量同步至Doris/SelectDB,主要是通过Flink计算引擎、基于Mysql CDC Connector和Doris社区研发的Flink Doris Connector 实现的。

无需提前建表,当Flink任务启动后,Flink Doris Connector 会自动识别对应的Doris表是否存在,不存在则自动创建Doris表,如果存在则直接启动同步任务。

并且能够支持Mysql源表Schema增删列自动变更至Doris,不需要手动变更Doris表的Schema或重启Job,便利之极。

前提概要

1. 安装部署

Doris/Flink/Mysql的安装部署流程可以根据版本自行Google。

① Doris下载地址:

https://doris.apache.org/zh-CN/download

② Flink下载地址: 

https://dlcdn.apache.org/flink/flink-1.17.1

③ MySQL包下载地址:

https://downloads.mysql.com/archives/installer

2. JAR包准备

① 数据源
Flink源端读取MySQL数据的连接器,可通过该地址下载flink-sql-connector-mysql-cdc-2.4.1.jar:
https://repo1.maven.org/maven2/com/ververica/flink-sql-connector-mysql-cdc/2.4.0/flink-sql-connector-mysql-cdc-2.4.0.jar
② 目标端
Flink目标端写Doris的Flink Doris Connector ,可通过该地址获取: 
https://cdnd.selectdb.com/zh-CN/docs/ecosystem/flink-doris-connector

3. 脚本模版

    ./bin/flink run \
    -Dexecution.checkpointing.interval=10s \
    -Dparallelism.default=1 \
    -c org.apache.doris.flink.tools.cdc.CdcTools \
    lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \
    mysql-sync-database \
    --database doris_db \
    --mysql-conf hostname= 127.0.0.1 \
    --mysql-conf port= 3306 \
    --mysql-conf username= root \
    --mysql-conf password= root \
    --mysql-conf database-name= mysql_db \
    --mysql-conf scan.startup.mode=initial \
    --including-tables "test.*" \
    --sink-conf fenodes= 127.0.0.1:8030 \
    --sink-conf username=root \
    --sink-conf password=root \
    --sink-conf jdbc-url=jdbc:mysql://127.0.0.1:9030 \
    --sink-conf sink.label-prefix=label \
        --table-conf replication_num=1 \
    • –job-name Flink任务名称, 非必需。

    • -Dexecution.checkpointing.interval checkpoint时间间隔,实际task中是需要checkpoint后才执行数据写入。

    • –database 同步到Doris的数据库名。

    • –mysql-conf MySQL CDC Source 配置,

      例如–mysql-conf hostname=127.0.0.1 ,可以在这里查看所有配置MySQL-CDC:

      https://ververica.github.io/flink-cdc-connectors/release-2.4/content/connectors/mysql-cdc.html

      其中hostname/username/password/database-name 是必需的。

    • –including-tables 需要同步的MySQL表,可以使用"|" 分隔多个表,并支持正则表达式。比如–including-tables “test.*” 就是同步所有以test开头的表。

    • –sink-conf Doris Sink 的所有配置,可以在这里查看完整的配置项:

      https://doris.apache.org/zh-CN/docs/dev/ecosystem/flink-doris-connector/#%E9%80%9A%E7%94%A8%E9%85%8D%E7%BD%AE%E9%A1%B9

    • –table-conf Doris表的配置项,即properties中包含的内容。例如 --table-conf replication_num=1

    快速体验

    1. 单表同步

    例如单表同步MySQL中Star Schema Benchmark的part零件信息表,共1600000条记录。

      ./bin/flink run \
      -Dexecution.checkpointing.interval=10s \
      -Dparallelism.default=1 \
      -c org.apache.doris.flink.tools.cdc.CdcTools \
      lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \
      mysql-sync-database \
      --database ssb_test \
      --mysql-conf hostname=172.21.16.12 \
      --mysql-conf port=3306 \
      --mysql-conf username=root \
      --mysql-conf password=root \
      --mysql-conf database-name=ssb_test \
      --mysql-conf scan.startup.mode=initial \
      --mysql-conf scan.incremental.snapshot.chunk.key-column=ssb_test.part:p_partkey \
      --including-tables "part" \
      --sink-conf fenodes=127.0.0.0:8030 \
      --sink-conf username=root \
      --sink-conf password=root \
      --sink-conf jdbc-url=jdbc:mysql://127.0.0.0:9030 \
      --sink-conf sink.label-prefix=label \
          --table-conf replication_num=1 \

      在FLINK_HOME执行上述命令后,WEB UI中会新增相应Job。

      查看taskmanager日志可以发现,task会先对mysql主键做切分,然后再根据ckp的时间间隔按批写入commit(默认2pc提交)。

      默认环境10秒ckp下,单并发写入只需3min左右即可同步完160W数据,且单副本数据压缩率高达90%

      2. 整库同步

      (1)数据准备
      基于【单表同步】的part零件信息表,新建一个db、然后copy几张相同表进行整库同步体验。
        -- 创建测试库
        create database ssb_test_part_all;


        -- 创建测试表 part01~3
        CREATE TABLE `part03` (
        `p_partkey` int(11) DEFAULT NULL,
        `p_name` varchar(23) DEFAULT NULL,
        `p_mfgr` varchar(7) DEFAULT NULL,
        `p_category` varchar(8) DEFAULT NULL,
        `p_brand` varchar(10) DEFAULT NULL,
        `p_color` varchar(12) DEFAULT NULL,
        `p_type` varchar(26) DEFAULT NULL,
        `p_size` int(11) DEFAULT NULL,
        `p_container` varchar(11) DEFAULT NULL
        ) ENGINE=InnoDB DEFAULT CHARSET=utf8




        -- 数据写入 part01~3
        insert into ssb_test_part_all.part01
        select * from ssb_test.part
        数据准备完成,如下图所示。
        (2)整库体验
        整库同步MySQL中ssb_test_part_all库中的所有表,共1600000*3条记录。
          ./bin/flink run \
          -Dexecution.checkpointing.interval=10s \
          -Dparallelism.default=1 \
          -c org.apache.doris.flink.tools.cdc.CdcTools \
          lib/flink-doris-connector-1.17-1.5.0-SNAPSHOT.jar \
          mysql-sync-database \
          --database ssb_test_part_all \
          --mysql-conf hostname=127.0.0.0.12 \
          --mysql-conf port=3306 \
          --mysql-conf username=root \
          --mysql-conf password=root \
          --mysql-conf database-name=ssb_test_part_all \
          --mysql-conf scan.startup.mode=initial \
          --mysql-conf scan.incremental.snapshot.chunk.key-column=ssb_test_part_all.part01:p_partkey,ssb_test_part_all.part02:p_partkey,ssb_test_part_all.part03:p_partkey \
          --including-tables ".*" \
          --sink-conf fenodes=127.0.0.0:8030 \
          --sink-conf username=root \
          --sink-conf password=root \
          --sink-conf jdbc-url=jdbc:mysql://127.0.0.0:9030 \
          --sink-conf sink.label-prefix=label \
              --table-conf replication_num=1 \
          在FLINK_HOME执行上述命令后,WEB UI中会新增相应整库同步Job。
          同步完成结果比对,不仅简易高效,且单副本数据压缩率高达90%+
          常见问题

          1. MySQL通信异常

          可能是因为mysql信息填写错误或驱动包不兼容导致。

          2. MySQL无Key表同步异常

          mysql源表没有设置主键,可以在脚本中指定–mysql-conf scan.incremental.snapshot.chunk.key-column=database.table:column,database.table1.column… 解决。

          3. CKP ON HDFS 连接异常

          使用hadoop作为checkpoint时报如下异常:

            Caused by: java.net.ConnectException: 
            Call From hadoop1/ip to hadoop1:8020 
            failed on connection exception
            java.net.ConnectException: Connection refused; 
            For more details see:  http://wiki.apache.org/hadoop/ConnectionRefused

            这类异常一般是由于网络原因或端口配置错误导致,可以先进行telnet ip port看是否通,再进行调整处理。

            4. CKP ON HDFS 权限异常

            使用默认root跑脚本时,如果出现如下异常:

              Caused by: org.apache.hadoop.security.AccessControlException: 
              Permission denied: user=root, access=WRITE, inode="/":hadoop:supergroup:drwxr-xr-x

              这类异常一般切换至对应用户或者授权即可,这个case可以切换至hadoop用户起 或 将root用户加到hdfs用户组中即可。

              5. 官方【Flink Doris Connector】QA

              其它问题可以参考 官方【Flink Doris Connector】QA 或 私 或 在开源社区群交流。

              至此,【Apache Doris】一键实现MySQL万表整库同步 分享结束,查阅过程中若遇到问题欢迎留言交流。

              最后,欢迎关注Flink Forward Asia 2023 FFA 2023 ,Doris Committer 吴迪分享:《Flink 到 Doris 实时写入实践:基于 Flink CDC 构建更实时高效的数据集成链路》

              往期推荐

              大数据平台开发规范示例

              【Apache Doris】Manager 极致丝滑地运维管理

              【Apache Doris】审计日志插件 | 快速体验

              Kafka TO Doris 保姆级入门详解

              Apache Doris 资源隔离详解

              Apache Doris IP变更问题详解

              ChatGPT快速入门

              如何正确地使用ChatGPT(角色扮演+提示工程)

              超强满血不收费的AI绘图教程来了(在线Stable Diffusion一键即用)

              文章转载自一臻数据,如果涉嫌侵权,请发送邮件至:contact@modb.pro进行举报,并提供相关证据,一经查实,墨天轮将立刻删除相关内容。

              评论