Skip to main content

Apache Doris Connector

Apache Doris Connector

Apache Doris is a high-performance, and real-time analytical database, which could support high-concurrent point query scenarios. Apache StreamPark encapsulates DoirsSink for writing data to Doris in real-time, based on its stream loads.

Write with Apache StreamPark™

DorisSink only supports JSON format (single-layer) writing currently, such as: {"id":1,"name":"streampark"} The example of the running program is Java, as follows:

Configuration list

doris.sink:
fenodes: 127.0.0.1:8030 # doris fe http url
database: test # doris database
table: test_tbl # doris table
user: root
password: 123456
batchSize: 100 # doris sink batch size per streamload
intervalMs: 3000 # doris sink the time interval of each streamload
maxRetries: 1 # stream load retries
streamLoad: # doris streamload own parameters
format: json
strip_outer_array: true
max_filter_ratio: 1

Write data to Doris

package org.apache.streampark.test.flink.java.datastream;

import org.apache.streampark.flink.core.StreamEnvConfig;
import org.apache.streampark.flink.core.java.sink.doris.DorisSink;
import org.apache.streampark.flink.core.java.source.KafkaSource;
import org.apache.streampark.flink.core.scala.StreamingContext;
import org.apache.streampark.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();
}
}

hint

It is recommended to set batchSize to insert data in batches to improve performance. At the same time, to improve real-time performance, intervalMs is supported for batch writing. The number of streamload retries can be increased by setting maxRetries.