Apache Doris 是一款基于大规模并行处理技术的分布式 SQL 数据库，主要面向 OLAP 场景。 Apache StreamPark 基于 Doris 的 stream load 封装了 DoirsSink 用于向 Doris 实时写入数据。

Apache StreamPark™ 方式写入

目前 DorisSink 只支持 JSON 格式（单层）写入，如 {"id":1,"name":"streampark"}。

示例程序是 Java 程序，具体如下。

配置信息

  1. doris.sink:
  2.   fenodes:  127.0.0.1:8030    # doris fe http 请求地址
  3.   database: test              # doris database
  4.   table: test_tbl             # doris table
  5.   user: root
  6.   password: 123456
  7.   batchSize: 100         # doris sink 每次 streamload 的批次大小
  8.   intervalMs: 3000       # doris sink 每次 streamload 的时间间隔
  9.   maxRetries: 1          # stream load 的重试次数
  10.   streamLoad:            # doris streamload 自身的参数
  11.     format: json
  12.     strip_outer_array: true
  13.     max_filter_ratio: 1

写入 Doris

Java

  1. package org.apache.streampark.test.flink.java.datastream;
  3. import org.apache.streampark.flink.core.StreamEnvConfig;
  4. import org.apache.streampark.flink.core.java.sink.doris.DorisSink;
  5. import org.apache.streampark.flink.core.java.source.KafkaSource;
  6. import org.apache.streampark.flink.core.scala.StreamingContext;
  7. import org.apache.streampark.flink.core.scala.source.KafkaRecord;
  8. import org.apache.flink.api.common.functions.MapFunction;
  9. import org.apache.flink.streaming.api.datastream.DataStream;
  11. public class DorisJavaApp {
  13.     public static void main(String[] args) {
  14.         StreamEnvConfig envConfig = new StreamEnvConfig(args, null);
  15.         StreamingContext context = new StreamingContext(envConfig);
  16.         DataStream<String> source = new KafkaSource<String>(context)
  17.             .getDataStream()
  18.             .map((MapFunction<KafkaRecord<String>, String>) KafkaRecord::value)
  19.             .returns(String.class);
  21.         new DorisSink<String>(context).sink(source);
  23.         context.start();
  24.     }
  25. }

建议设置 batchSize 来批量插入数据提高性能,同时为了提高实时性，支持间隔时间 intervalMs 来批次写入

可以通过设置 maxRetries 来增加streamload的重试次数。