Flink向Doris表写入数据（Sink）

业务场景

最近在工作中遇到了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方法中，需要注意以下几个地方：

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

关于doris中的类型和Flink类型之间的对应关系，可以参考官档Flink Doris Connector - Apache Doris

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

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