经过各种实践，发现比较适合中小公司的方式。分为全量和增量。
1、参考链接：
1、Flink Doris Connector：https://doris.apache.org/zh-CN/docs/dev/ecosystem/flink-doris-connector
2、JSON格式数据导入：https://doris.apache.org/zh-CN/docs/dev/data-operate/import/import-way/load-json-format
2、全量导入
通过外部表同步数据
3、增量导入
1、dataocean监听kafka然后insert，这种方式，官方不建议，大量的insert会把doris-be打崩溃。（目前正在用的方式）。
2、binlog load的方式，以前旧版本尝试有问题，使用新版本尝试看看。经过新版本的尝试，发现这种方式不适合，详见：binlog方式调研。
3、Flink CDC，但是这种方式得学习和搭建Flink，使得成本变高了。准备尝试这种方式了，后期使用这种方式。
4、Flink的方式
1、订阅kafka的方式
canal发个kafka的消息，可以通过如下方式同步给doris。
但是问题是：无法控制同步哪些数据到哪个表
CREATE ROUTINE LOAD ocean.test1 ON cloud_uc_company
COLUMNS(id, name, delete_at)
PROPERTIES
(
"desired_concurrent_number"="3",
"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200",
"strict_mode" = "false",
"format" = "json",
"jsonpaths" = "[\"$.id\",\"$.name\",\"$.delete_at\"]",
"strip_outer_array" = "true",
"json_root" = "$.data"
)
FROM KAFKA
(
"kafka_broker_list" = "10.20.1.26:9092",
"kafka_topic" = "topic-periodfour",
'property.group.id' = 'doris_ods_log',
'property.client.id' = 'doris_ods_log',
'property.kafka_default_offsets' = 'OFFSET_BEGINNING'
);
2、Flink代码可以拿到kafka消息
2.1、原理：
这种方式是拿到canal发送给里的kafka消json消息，然后取出其中的data部分，然后再写到doris里去。这种方式其实是通过doris的json序列化器，将data部分序列化成json字符串，然后再写入doris。
2.2、碰到的问题：
1、问题描述：
DorisRuntimeException: stream load error: [ANALYSIS_ERROR]errCode = 2, detailMessage = current running txns on db 11008 is 100, larger than limit 100, see more in null
解决方式：
这是因为同一个库并发导入超过了100，可通过调整 fe.conf的参数
max_running_txn_num_per_db来解决。
执行sql：admin set frontend config ('max_running_txn_num_per_db' = '1000');
2、问题描述：
DorisRuntimeException: stream load error: [INTERNAL_ERROR]too many filtered rows, see more in http://10.20.0.92:8040/api/_load_error_log?file=__shard_182/error_log_insert_stmt_8f4f21b2fa64f604-8fbb0ec0fba8129d_8f4f21b2fa64f604_8fbb0ec0fba8129d。
打开链接，看到了错误信息：Reason: JSON data is array-object,strip_outer_arraymust be TRUE。
解决方式：
需要增加代码：设置pro.setProperty("strip_outer_array", "true");
2.3、代码：
package com.bm001.flinkdemo;
import com.google.gson.Gson;
import org.apache.commons.collections.CollectionUtils;
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.DorisRecordSerializer;
import org.apache.doris.flink.sink.writer.JsonDebeziumSchemaSerializer;
import org.apache.doris.flink.sink.writer.RowDataSerializer;
import org.apache.doris.flink.sink.writer.SimpleStringSerializer;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.types.DataType;
import java.io.IOException;
import java.time.LocalDate;
import java.util.List;
import java.util.Map;
import java.util.Properties;
import java.util.UUID;
/**
* 这种方法是将kafka的json消息，获取其中的data部分，
* 然后再通过doris的json序列化器，将data部分序列化成json字符串，然后再写入doris
* <p>
* 每个同步的表都需要增加一个任务
*/
public class Kafka2DorisDataStreamDemo {
public static void main(String[] args) throws Exception {
Properties props = new Properties();
props.put("bootstrap.servers", "10.20.1.26:9092");
props.put("group.id", "group-flinkdemo");
props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
props.put("auto.offset.reset", "latest");
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.enableCheckpointing(1000);
env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
//source config
FlinkKafkaConsumer<String> flinkKafkaConsumer = new FlinkKafkaConsumer<>("topic-periodfour", new SimpleStringSchema(), props);
//sink config
DorisSink.Builder<String> builder = DorisSink.builder();
DorisOptions.Builder dorisBuilder = DorisOptions.builder();
DorisOptions dorisOptions = dorisBuilder.setFenodes("10.20.0.91:8030")
.setTableIdentifier("ocean.cloud_uc_company")
.setUsername("root")
.setPassword("root12345")
.build();
Properties pro = new Properties();
//json data format
pro.setProperty("format", "json");
pro.setProperty("read_json_by_line", "true");
pro.setProperty("strip_outer_array", "true"); //如果serialize返回的是json数组格式，此处必须要设置为true
DorisExecutionOptions executionOptions = DorisExecutionOptions.builder().setLabelPrefix("label-doris" + UUID.randomUUID()) //streamload label prefix,
.setStreamLoadProp(pro).build();
builder.setDorisReadOptions(DorisReadOptions.builder().build()).setDorisExecutionOptions(executionOptions)
// .setSerializer(new SimpleStringSerializer()) //serialize according to string
.setSerializer(new DorisRecordSerializer<String>() {
@Override
public byte[] serialize(String s) throws IOException {
CanalMessage canalMessage = new Gson().fromJson(s, CanalMessage.class);
String type = canalMessage.getType();
if ("INSERT".equals(type) || "UPDATE".equals(type)) {
if ("uc_company".equals(canalMessage.getTable())) {
// Map<String, Object> dataMap = canalMessage.getData().get(0);
// return new Gson().toJson(dataMap).getBytes();
return new Gson().toJson(canalMessage.getData()).getBytes();
}
}
return null;
}
}).setDorisOptions(dorisOptions);
//build stream
DataStreamSource<String> dataStreamSource = env.addSource(flinkKafkaConsumer);
dataStreamSource.sinkTo(builder.build());
env.execute("flink kafka to doris by datastream");
}
}
