Skip to main content

Doris Connector

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

StreamX 方式写入

StreamX写入 doris的数据, 目前DorisSink只支持JSON格式(单层)写入,如:{"id":1,"name":"streamx"} 运行程序样例为java,如下:

配置信息

doris.sink:
fenodes: 127.0.0.1:8030 //doris fe http 请求地址
database: test //doris database
table: test_tbl //doris table
user: root
password: 123456
batchSize: 100 //doris sink 每次streamload的批次大小
intervalMs: 3000 //doris sink 每次streamload的时间间隔
maxRetries: 1 //stream load的重试次数
streamLoad: //doris streamload 自身的参数
format: json
strip_outer_array: true
max_filter_ratio: 1

写入doris

package com.streamxhub.streamx.test.flink.java.datastream;

import com.streamxhub.streamx.flink.core.StreamEnvConfig;
import com.streamxhub.streamx.flink.core.java.sink.doris.DorisSink;
import com.streamxhub.streamx.flink.core.java.source.KafkaSource;
import com.streamxhub.streamx.flink.core.scala.StreamingContext;
import com.streamxhub.streamx.flink.core.scala.source.KafkaRecord;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;

public class DorisJavaApp {

public static void main(String[] args) {
StreamEnvConfig envConfig = new StreamEnvConfig(args, null);
StreamingContext context = new StreamingContext(envConfig);
DataStream<String> source = new KafkaSource<String>(context)
.getDataStream()
.map((MapFunction<KafkaRecord<String>, String>) KafkaRecord::value)
.returns(String.class);

new DorisSink<String>(context).sink(source);

context.start();
}
}

提示

建议设置 batchSize 来批量插入数据提高性能,同时为了提高实时性,支持间隔时间 intervalMs 来批次写入
可以通过设置 maxRetries 来增加streamload的重试次数。