SeaTunnel流处理同步MySQL数据至ClickHouse

ClickHouse是一种OLAP类型的列式数据库管理系统，ClickHouse完美的实现了OLAP和列式数据库的优势，因此在大数据量的分析处理应用中ClickHouse表现很优秀。

SeaTunnel是一个分布式、高性能、易扩展、用于海量数据同步和转化的数据集成平台。用户只需要配置作业信息，就能完成数据的同步。提交作业后，源连接器负责并行读取数据并将数据发送到下游转换或直接发送到接收器，接收器将数据写入目标。

SeaTunnel任务配置及启动

MySQL CDC支持stream流处理模式，本示例使用默认启动模式 INITIAL，先同步历史数据，后增量同步。先将MySQL的test数据库下bigtest表中的10000条历史数据，同步到ClickHouse数据库下default.tests中，之后增量同步新数据至ClickHouse中。

MySQL建表，语句如下：

CREATE TABLE `bigtest` (
  `id` int(11) NOT NULL,
  `name` varchar(100) DEFAULT NULL,
  `quantity` int(11) DEFAULT NULL,
  PRIMARY KEY (`id`)
)

向MySQL中插入数据，格式如下：

insert into test.bigtest values(1,'banana',1);

ClickHouse建表，语句如下：

CREATE TABLE default.jdbc
(
    `id` Int32,
    `name` String,
    `quantity` Int32
)
ENGINE = MergeTree
ORDER BY id

1、启动MySQL binlog

修改配置文件参数启动binlog

vi /etc/my.cnf

添加如下内容：

[mysqld]

character-set-server=utf8

log-bin=mysql-bin

binlog-format=ROW

server_id=1

之后重启MySQL服务

2、下载jdbc和连接器

下载MySQLjdbc并放至 '$SEATNUNNEL_HOME/plugins/jdbc/'目录下

编辑plugin_config：

vi seatunnel-2.x.x/config/plugin_config

添加所需要的connector如：

--connectors-v2--

connector-cdc-mysql

connector-clickhouse

connector-fake

connector-console

--end--

执行以下命令下载所需插件（2.3.3即下载的插件版本）：

sh bin/install-plugin.sh 2.3.3

调用安装脚本的时候会在maven的中央仓库下载对应的jar包，下载较慢，可以手动下载。

手动下载连接器：

下载连接器connector-cdc-mysql-2.x.x.jar，并放至'$SEATNUNNEL_HOME/connectors/seatunnel/'目录下。

连接器下载地址：

https://repo.maven.apache.org/maven2/org/apache/seatunnel/

3、编写配置文件

在'$SEATNUNNEL_HOME/config'目录下，新建配置文件

配置文件内容示例如下：

env {
  execution.parallelism = 1
  job.mode = "STREAMING"
  checkpoint.interval = 2000
}
source {
  MySQL-CDC {
    base-url = "jdbc:mysql://localhost:3306/test"
    username = "root"
    password = "123456"
    table-names = ["test.test"]
  }
}
sink {
  Clickhouse {
    host = "localhost:8123"
    database = "default"
    table = "tests"
    username = "default"
    password = "123456"
  }
}

4、启动任务

在'$SEATNUNNEL_HOME'目录下，使用启动命令：

./bin/seatunnel.sh --config ./config/mysqlcdctock.template -e local

此命令将以 local (本地模式) 运行您的SeaTunnel作业。由于是流处理模式，任务会一直执行。

进入ClickHouse查看数据写入情况select * from default.tests，可以看到10000条历史数据已经同步至ClickHouse了。