概述

本文主要讲述的是通过flink CDC +Dinky 同步mysql数据到paimon中自动建表的实践。Dinky通过定义CDCSOURCE语法，可以直接自动构建一个整库入仓入湖的实时任务，避免了大量的数据库连接和DDL编写，同时采用多source合并的优化策略，减少了同一作业中的source数量，避免了Binlog的重复读取，从而减轻了源库的压力。从而减轻了源库的压力。CDCSOURCE 语句用于将上游指定数据库的所有表的数据采用一个任务同步到下游系统。

前置条件

安装好Dinky
安装好flink Ha集群

环境准备

相关jar包

flink-sql-connector-mysql-cdc-3.2.1.jar
mysql-connector-java-8.0.30.jar
paimon-flink-1.19-0.9.0.jar
flink-shaded-hadoop-2-uber-2.8.3-10.0.jar
flink-cdc-dist-3.2.1.jar

同步任务

EXECUTE CDCSOURCE dinky_paimon_test
WITH
  (
    'connector' = 'mysql-cdc',
    'hostname' = 'ip',
    'port' = '3306',
    'username' = 'root',
    'password' = 'xxx',
    'checkpoint' = '10000',
    'parallelism' = '2',
    'scan.startup.mode' = 'initial',
    'database-name' = 'lzs-ad',
    'table-name' = 'lzs-ad\.t_report',
    'server-id'='5400-5499',
    'sink.connector' = 'paimon',
    'sink.path' = 'hdfs://mycluster/user/data/paimon/#{schemaName}.db/#{tableName}',
    'sink.auto-create' = 'true',
    'sink.bucket'='4'
  );

错误1:

Caused by: java.util.ServiceConfigurationError: org.dinky.cdc.SinkBuilder: org.dinky.cdc.doris.DorisSinkBuilder Unable to get public no-arg constructor

解决：添加flink-doris-connector jar包。

错误2:

解决：

在执行前设置：set table.exec.sink.upsert-materialize = NONE;

修改后的完整sql代码：

set table.exec.sink.upsert-materialize = NONE;  
set table.exec.sink.parquet.snapshot.time-retained = 1d ; 
EXECUTE CDCSOURCE dinky_paimon_test
WITH
  (
    'connector' = 'mysql-cdc',
    'hostname' = 'ip',
    'port' = '3306',
    'username' = 'root',
    'password' = 'xxx!',
    'checkpoint' = '10000',
    'parallelism' = '2',
    'scan.startup.mode' = 'initial',
    'database-name' = 'lzs-ad',
    'table-name' = 'lzs-ad\.t_report',
    'server-id'='5400-5499',
    'sink.connector' = 'paimon',
    'sink.path' = 'hdfs://mycluster/user/data/paimon/#{schemaName}.db/#{tableName}',
    'sink.auto-create' = 'true',
    'sink.bucket'='4'
  );

StarRocks 查询Paimon表

CREATE EXTERNAL CATALOG paimon_catalog_fs
PROPERTIES
(
    "type" = "paimon",
    "paimon.catalog.type" = "filesystem",
    "paimon.catalog.warehouse" = "hdfs://bigdata02:9000/user/data/paimon"
);

--切换数据CATALOG
set CATALOG paimon_catalog_fs; 
use paimon_catalog_fs;

-- 删除catalog
drop CATALOG paimon_catalog_fs;

SHOW DATABASES FROM paimon_catalog_fs;

USE paimon_catalog_fs.lzs-ad;
select * from t_report limit 3;