在本文中，我们将探讨如何使用Flink CDC将数据接入Doris数据库。
Flink CDC是一种开源的流式数据变更捕获工具，可以捕获数据源中的变更并将其传递到目标系统中。
Apache Doris数据库是一个开源的分布式列式存储系统，用于高效地存储和查询海量数据。
完整获取代码
由于篇幅限制，文中代码部分只展示了核心代码。我将本文中涉及的完整代码打包放在我的百度网盘中了，下载下来之后，导入idea中，找到MysqlCDC2DorisJob.java运行 main 函数即可。
有需要的小伙伴可以关注我的公众号，并在后台回复关键字：cdc，获取分享链接和你的专属密码哦。
一、原理
通过Flink CDC将数据接入Doris，其实就是利用了Flink CDC的flink-connector-mysql-cdc插件和Doris的Stream Load插件。
flink-connector-mysql-cdc可以捕获mysql数据库变更的数据 Doris Stream Load是通过HTTP协议，将数据发送Doris FE或者Doris BE，如果发送到FE，FE会将数据进一步转发到BE。一般来说，为了减少FE的压力，我们建议直接将数据发送到BE。
我们可以通过轮询算法来选取指定的BE，也可以通过nginx的四层代理对BE节点做负载均衡。
二、Doris Stream Load介绍
Stream Load 是一个同步的导入方式，通过发送HTTP请求将本地文件或数据流导入到 Doris 中。如果提交到 FE，则 FE 会通过 HTTP redirect 指令将请求转发给某一个BE。也可以直接提交导入命令给某一个指定的 BE。Stream Load 同步执行导入并返回导入结果。我们可以通过请求的返回体判断本次导入是否成功。
Doris Stream load 主要用于导入本地文件，或导入实时数据流中的数据。
其实，在Doris中，导入数据的方式有很多，比如Broker Load，Mini Load，Spark Load，Hdfs Load，Routine Load，Binlog Load，S3 Load…，这些导入方式可以参考Doris官网文档中的数据导入一节，这里不做过多赘述。
下面介绍一下Stream Load的导入命令以及相关参数的含义，如果已经有所熟悉，跳过即可。
1. Stream Load导入命令
curl --location-trusted -u user:passwd [-H ""...] -T data.file -XPUT http://fe_host[或者be_host]:http_port/api/{db}/{table}/_stream_load
2. 参数解释
-u user:passwd
Stream Load 创建导入的协议使用的是 HTTP 协议，通过 Basic access authentication 进行签名。Doris 系统会根据签名验证用户身份和导入权限。-
-H key:value
Http请求的Header部分，格式为:-H "key:value"。我们可以设置以下参数：
3. 返回结果
{
"TxnId": 1001,
"Label": "a6f3bc78-012c-45d9-9e4c-faa0a0149bae",
"Status": "Success",
"ExistingJobStatus": "FINISHED", // optional
"Message": "OK",
"NumberTotalRows": 10,
"NumberLoadedRows": 20,
"NumberFilteredRows": 1,
"NumberUnselectedRows": 0,
"LoadBytes": 1231231,
"LoadTimeMs": 800,
"BeginTxnTimeMs": 1,
"StreamLoadPutTimeMs": 2,
"ReadDataTimeMs": 325,
"WriteDataTimeMs": 1933,
"CommitAndPublishTimeMs": 106,
"ErrorURL": "http://192.168.1.1:8042/api/_load_error_log?file=__shard_0/error_log_insert_stmt_db18266d4d9b4ee5-abb00ddd64bdf005_db18266d4d9b4ee5_abb00ddd64bdf005"
}
4. 取消导入任务
5. 查看 Stream Load
show stream load 命令来查看已经完成的 stream load 任务。enable_stream_load_record=true三、代码实践
1. 准备工作
mysql表
CREATE TABLE `t_ds_user` (
`id` int NOT NULL AUTO_INCREMENT COMMENT 'user id',
`user_name` varchar(64) DEFAULT NULL COMMENT 'user name',
`user_password` varchar(64) DEFAULT NULL COMMENT 'user password',
`user_type` tinyint DEFAULT NULL COMMENT 'user type, 0:administrator，1:ordinary user',
`email` varchar(64) DEFAULT NULL COMMENT 'email',
`phone` varchar(11) DEFAULT NULL COMMENT 'phone',
`tenant_id` int DEFAULT NULL COMMENT 'tenant id',
`create_time` datetime DEFAULT NULL COMMENT 'create time',
`update_time` datetime DEFAULT NULL COMMENT 'update time',
`queue` varchar(64) DEFAULT NULL COMMENT 'queue',
`state` tinyint DEFAULT '1' COMMENT 'state 0:disable 1:enable',
`time_zone` varchar(32) DEFAULT NULL COMMENT 'time zone',
PRIMARY KEY (`id`),
UNIQUE KEY `user_name_unique` (`user_name`)
) ENGINE=InnoDB AUTO_INCREMENT=3 DEFAULT CHARSET=utf8mb4
mysql表初始化数据
INSERT INTO t_ds_user (user_name,user_password,user_type,email,phone,tenant_id,create_time,update_time,queue,state,time_zone) VALUES
('admin','7ad2410b2f4c074479a8937a28a22b8f',0,'xxx@qq.com','',0,'2018-03-27 15:48:50','2018-10-24 17:40:22',NULL,1,NULL);
Doris 表
CREATE TABLE `t_ds_user` (
`id` int(11) NOT NULL COMMENT "user id",
`user_name` varchar(64) NULL COMMENT "user name",
`user_password` varchar(64) NULL COMMENT "user password",
`user_type` tinyint(4) NULL COMMENT "user type, 0:administrator，1:ordinary user",
`email` varchar(64) NULL COMMENT "email",
`phone` varchar(11) NULL COMMENT "phone"
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "用户表"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 3",
"in_memory" = "false",
"storage_format" = "V2"
);
2. 引入相关jar包
<!-- flink -->
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-java</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-streaming-java_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-clients_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<dependency>
<groupId>org.apache.hadoop</groupId>
<artifactId>hadoop-client</artifactId>
<version>3.1.3</version>
</dependency>
<dependency>
<groupId>org.apache.flink</groupId>
<artifactId>flink-connector-jdbc_2.12</artifactId>
<version>${flink.version}</version>
</dependency>
<!-- mysql驱动包 -->
<dependency>
<groupId>mysql</groupId>
<artifactId>mysql-connector-java</artifactId>
<version>5.1.49</version>
</dependency>
<!-- mysql cdc -->
<dependency>
<groupId>com.alibaba.ververica</groupId>
<artifactId>flink-connector-mysql-cdc</artifactId>
<version>1.3.0</version>
</dependency>
3. 编写flink主体程序
package com.eyesmoons.cdc.mysql;
import com.alibaba.ververica.cdc.connectors.mysql.MySQLSource;
import com.alibaba.ververica.cdc.connectors.mysql.table.StartupOptions;
import com.alibaba.ververica.cdc.debezium.DebeziumSourceFunction;
import com.eyesmoons.cdc.deserialization.CustomerDeserializationSchemaMysql;
import com.eyesmoons.cdc.sink.DorisSink;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
@Slf4j
public class MysqlCDC2DorisJob {
public static void main(String[] args) throws Exception {
// 1.创建执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 2.Flink-CDC 将读取 binlog 的位置信息以状态的方式保存在 CK,如果想要做到断点续传,需要从 Checkpoint 或者 Savepoint 启动程序
// 2.1 开启 Checkpoint,每隔 5 秒钟做一次 CK
// env.enableCheckpointing(5000L);
// 2.2 指定 CK 的一致性语义
// env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
// 2.3 设置任务关闭的时候保留最后一次 CK 数据
// env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
// 2.4 指定从 CK 自动重启策略
// env.setRestartStrategy(RestartStrategies.fixedDelayRestart(3, 2000L));
// 2.5 设置状态后端
// env.setStateBackend(new FsStateBackend("hdfs://hadoop102:8020/flinkCDC"));
// 2.6 设置访问 HDFS 的用户名
// System.setProperty("HADOOP_USER_NAME", "hdfs");
// 3.创建 Flink-MySQL-CDC 的 Source
DebeziumSourceFunction<String> mysqlSource = MySQLSource.<String>builder()
.hostname("127.0.0.1")
.port(3306)
.username("shengyu")
.password("123456")
.databaseList("dolphinscheduler")
// 可选配置项,如果不指定该参数,则会读取上一个配置下的所有表的数据，注意：指定的时候需要使用"db.table"的方式
.tableList("dolphinscheduler.t_ds_user")
// 指定数据读取位置：initial，latest-offset，timestamp，specific-offset
.startupOptions(StartupOptions.initial())
.deserializer(new CustomerDeserializationSchemaMysql())
.build();
// 4.使用 CDC Source 从 MySQL 读取数据
DataStreamSource<String> mysqlDS = env.addSource(mysqlSource);
// 5.使用streamLoad加载数据到doris
mysqlDS.addSink(new DorisSink());
// 6.执行任务
env.execute();
}
}
4. 自定义Doris Sink
package com.eyesmoons.cdc.sink;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.eyesmoons.cdc.utils.BeUtil;
import com.eyesmoons.cdc.utils.HttpClientUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.http.Header;
import org.apache.http.HttpStatus;
import org.apache.http.message.BasicHeader;
import java.nio.charset.StandardCharsets;
import java.util.*;
@Slf4j
public class DorisSink extends RichSinkFunction<String> {
// stream load请求地址
private static final String LOAD_URL = "http://%s/api/%s/%s/_stream_load";
// doris数据库
private static final String DORIS_DB_NAME = "demo";
// doris表名
private static final String DORIS_TABLE_NAME = "t_ds_user";
// doris数据库用户名
private static final String DORIS_USER_NAME = "shengyu";
// doris数据库密码
private static final String DORIS_PASSWORD = "123456";
// doris表字段
private static final String columns = "id,user_name,user_password,user_type,email,phone";
// json数据字段
private static final String jsonPaths = "[\"$.id\",\"$.user_name\",\"$.user_password\",\"$.user_type\",\"$.email\",\"$.phone\"]";
// doris be集合
private static final Map<String, Integer> beMap = new HashMap<>();
@Override
public void open(Configuration parameters) throws Exception {
super.open(parameters);
// 初始化be集合
beMap.put("doris_ip1:8040", 1);
beMap.put("doris_ip2:8040", 1);
beMap.put("doris_ip3:8040", 1);
}
@Override
public void invoke(String value, Context context) {
JSONObject json = JSON.parseObject(value);
JSONObject after = json.getJSONObject("after");
String data = after.toJSONString();
LoadResponse response = doStreamLoad(data);
System.out.println(JSON.toJSON(response));
}
/**
* 执行streamLoad
* @param data json数据
*/
public LoadResponse doStreamLoad(String data) {
String loadBe = BeUtil.getBeRoundRobin(beMap);
String beLoadUrl = String.format(LOAD_URL, loadBe, DORIS_DB_NAME, DORIS_TABLE_NAME);
log.info("stream load to be: {}", loadBe);
// 补全header信息
List<Header> headers = buildHeaders();
try {
// 执行post请求
String beResp = HttpClientUtil.putJson(beLoadUrl, data, headers);
beResp = beResp.replaceAll("\r|\n", "");
return new LoadResponse(HttpStatus.SC_OK, "ok", beResp, loadBe);
} catch (Exception e) {
String errorMessage = "failed to load data stream load: ";
log.error(errorMessage, e);
return new LoadResponse(-1, e.getMessage(), errorMessage, loadBe);
}
}
/**
* 添加header信息
*/
public List<Header> buildHeaders() {
List<Header> headers = new ArrayList<>();
String authEncoding = Base64.getEncoder().encodeToString(String.format("%s:%s", DORIS_USER_NAME, DORIS_PASSWORD).getBytes(StandardCharsets.UTF_8));
headers.add(new BasicHeader("Authorization", "Basic " + authEncoding));
headers.add(new BasicHeader("Expect", "100-continue"));
headers.add(new BasicHeader("strip_outer_array", "false"));
headers.add(new BasicHeader("format", "json"));
headers.add(new BasicHeader("columns", columns));
headers.add(new BasicHeader("jsonpaths", jsonPaths));
headers.add(new BasicHeader("strict_mode", "true"));
headers.add(new BasicHeader("label", buildLabel()));
return headers;
}
/**
* 生成stream load，必须唯一，所以此处用了UUID
*/
public String buildLabel() {
return String.format("label_%s_%s", DORIS_TABLE_NAME, UUID.randomUUID().toString().replaceAll("-", ""));
}
}
四、总结
cdc，获取完整代码的分享链接和你的专属密码哦。由于Doris Stream Load是基于HTTP协议的，所以我们在程序中更容易集成进来。 相比于其他数据采集工具，Flink CDC 和 Flink 的兼容性更好，以及更轻量级。而且使用Flink CDC，我们可以指定数据采集的位点、全量和增量采集，可以满足我们对数据采集的大部分需求。
原文作者：毛毛小妖的笔记；如有侵权，请联系管理员删除。
请登录后发表评论
登录
全部评论
