赞
踩
最近在工作中遇到了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方法中,需要注意以下几个地方:
关于doris中的类型和Flink类型之间的对应关系,可以参考官档Flink Doris Connector - Apache Doris
Doris官方提供的doris sink连接器,默认是已经开启了两阶段提交,不需要我们额外的过多设置。
以上就是Flink将数据最终写入Doris的大概主体代码,目前本人程序运行正常,各位可自行参考修改,有任何问题也欢迎评论区留言讨论。
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。