业务场景
最近在工作中遇到了Flink处理kafka中的数据，最后写入Doris存储的场景。

Apache Doris 是一款基于 MPP 架构的高性能、实时的分析型数据库，以高效、简单、统一的特点被人们所熟知，仅需亚秒级响应时间即可返回海量数据下的查询结果，不仅可以支持高并发的点查询场景，也能支持高吞吐的复杂分析场景。基于此，Apache Doris 能够较好的满足报表分析、即席查询、统一数仓构建、数据湖联邦查询加速等使用场景，用户可以在此之上构建大屏看板、用户行为分析、AB 实验平台、日志检索分析、用户画像分析、订单分析等应用。

以上介绍来自于Doris的官网，关于Doris的内容不做过多的介绍，可以参考官档 快速体验 Apache Doris - Apache Doris

依赖
本人是通过datastream的方式向Doris写入数据，并非通过FlinkSQL的方式，不过不管使用哪种方式，都需要以下的依赖

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

代码实现
在doris的官档中，数据通过datastream写入doris，支持两种不同的序列化方法，一种是String 数据流 (SimpleStringSerializer)，另一种是RowData 数据流 (RowDataSerializer)。本人使用的是后者，因为后者可以很好的兼容json类型的数据。

现将主类的业务代码提供如下，重点处通过注释做了说明

package cn.gwm.dp.main;

import cn.gwm.dp.consts.Constant;
import cn.gwm.dp.entity.TaskDetail;
import cn.gwm.dp.entity.TaskSum;
import cn.gwm.dp.entity.YRcanbus;
import cn.gwm.dp.functions.KafkaFunction;
import cn.gwm.dp.functions.KeyedFunction;
import cn.gwm.dp.functions.RowDataFunction;
import cn.gwm.dp.functions.TosFunction;
import org.apache.doris.flink.cfg.DorisExecutionOptions;
import org.apache.doris.flink.cfg.DorisOptions;
import org.apache.doris.flink.cfg.DorisReadOptions;
import org.apache.doris.flink.sink.DorisSink;
import org.apache.doris.flink.sink.writer.serializer.RowDataSerializer;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.types.DataType;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.config.SaslConfigs;

import java.util.Properties;

/**
* @Author: Spring
* @Description:
* @Date: Created on 10:27 2024/5/13
*/
public class NewbieTaskApp {
public static void main(String[] args) throws Exception {

StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 业务处理代码省略。。。

// 获得Doris Sink
DorisSink<RowData> taskDetailSink = getTaskDetailSink(parameterTool);

// 数据分流写入到doris
SideOutputDataStream<RowData> taskDetailData = eachRowData.getSideOutput(taskDetailOutputTag);
taskDetailData.sinkTo(taskDetailSink).name("taskDetailSink");

env.execute("newbie_task_online");
}

private static DorisSink<RowData> getTaskDetailSink(ParameterTool parameterTool) {
// doris的连接地址
String fenodes = parameterTool.get(Constant.FENODES);
// doris的表
String dorisDetailTable = parameterTool.get(Constant.DORIS_TASK_DETAIL_TABLE);
// doris连接用户
String dorisUser = parameterTool.get(Constant.DORIS_USER);
// doris连接密码
String dorisPwd = parameterTool.get(Constant.DORIS_PWD);

DorisSink.Builder<RowData> builder = DorisSink.builder();
DorisOptions.Builder optionsBuilder = DorisOptions.builder();
optionsBuilder.setFenodes(fenodes)
.setTableIdentifier(dorisDetailTable)
.setUsername(dorisUser)
.setPassword(dorisPwd);

Properties properties = new Properties();
// 指定处理json类型数据
properties.setProperty("format", "json");
properties.setProperty("read_json_by_line", "true");
DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();
// 这里设定的prefix，每个Flink应用都不能相同
executionBuilder.setLabelPrefix("label-task-detail-test2")
.setDeletable(false)
.setStreamLoadProp(properties);

// 指定要落入的doris表的字段
String[] fields = {"day", "uin", "platform", "task_type", "trip_id", "vin", "task_time",
"task_distance", "tja_ica_mod_disp", "noh_sts", "road_class", "create_time"};
// 指定doris表字段类型所对应的Flink的类型
DataType[] types = {DataTypes.DATE(),
DataTypes.CHAR(40),
DataTypes.TINYINT(),
DataTypes.TINYINT(),
DataTypes.INT(),
DataTypes.VARCHAR(64),
DataTypes.INT(),
DataTypes.FLOAT(),
DataTypes.TINYINT(),
DataTypes.TINYINT(),
DataTypes.TINYINT(),
DataTypes.TIMESTAMP()};

// 构建sink
builder.setDorisReadOptions(DorisReadOptions.builder().build())
.setDorisExecutionOptions(executionBuilder.build())
.setSerializer(RowDataSerializer.builder()
.setFieldNames(fields)
.setType("json")
.setFieldType(types).build())
.setDorisOptions(optionsBuilder.build());

return builder.build();
}
}

上述代码中，数据的业务处理逻辑我省略了，各位换成自己的。

在写入doris之前，我是通过分流的方式拿到要写入的数据，且数据的类型是RowData。各位的数据可根据自己的业务逻辑准备，如果你要写入doris的数据和我一样，也是json类型的，那么一定要使用RowData类型。

关于RowData类型的构建，可以简单的参考下面的代码

// TaskDetail 是一个自定义的实体类
TaskDetail taskDetail = tp.f0;
GenericRowData taskDetailRowData = new GenericRowData(10);
taskDetailRowData.setField(0, Integer.parseInt(String.valueOf(taskDetail.getDay().toEpochDay())));
taskDetailRowData.setField(1, StringData.fromString(taskDetail.getUin()));
taskDetailRowData.setField(2, (byte) taskDetail.getPlatform());
taskDetailRowData.setField(3, (byte) taskDetail.getTaskType());
taskDetailRowData.setField(4, taskDetail.getTripId());
taskDetailRowData.setField(5, StringData.fromString(taskDetail.getVin()));
taskDetailRowData.setField(6, taskDetail.getTaskTime());
taskDetailRowData.setField(7, taskDetail.getTaskDistance());
taskDetailRowData.setField(8, (byte) taskDetail.getTjaIcaModDisp());
taskDetailRowData.setField(9, (byte) taskDetail.getNohSts());

在获得doris sink的getTaskDetailSink方法中，需要注意以下几个地方：

fields 和 types 的指定，一定要和具体的doris表关联起来
doris中的数据类型要和Flink中的数据类型关联正确
setLabelPrefix 方法指定的标签，每个Flink应用都不一样。如果你的程序启动后，没有挂，且已经将部分数据写入到了doris中，那么在程序关停的时候，一定要设置savepoint路径，不能直接暴力的cancel，并且下次启动时从savepoint启动，否则程序会报错。如果你cancel了，那么下次启动程序前，必须要修改这个标签的值为不同的，才能启动成功。
关于doris中的类型和Flink类型之间的对应关系，可以参考官档Flink Doris Connector - Apache Doris

Doris官方提供的doris sink连接器，默认是已经开启了两阶段提交，不需要我们额外的过多设置。

以上就是Flink将数据最终写入Doris的大概主体代码，目前本人程序运行正常，各位可自行参考修改，有任何问题也欢迎评论区留言讨论。

本文系转载，版权归原作者所有，

转载自CSDN 程序吴老二 ，如若侵权请联系我们进行删除！  

《行业指标体系白皮书》下载地址：https://www.dtstack.com/resources/1057/?src=bbs

《数据治理行业实践白皮书》下载地址：https://www.dtstack.com/resources/1001/?src=bbs

《数栈V6.0产品白皮书》下载地址：https://www.dtstack.com/resources/1004/?src=bbs

想了解或咨询更多有关袋鼠云大数据产品、行业解决方案、客户案例的朋友，浏览袋鼠云官网：https://www.dtstack.com/?src=bbs

同时，欢迎对大数据开源项目有兴趣的同学加入「袋鼠云开源框架钉钉技术群」，交流最新开源技术信息，群号码：30537511，项目地址：https://github.com/DTStack