腾讯云数据仓库 TCHouse-D Flink Connector（Flink 实时或批量数据）

一、基于 Flink Connector 实现数据实时或批量导入 Doris

注意：这篇档适用于 flink-doris-connector 1.1.0之后的版本。

二、版本兼容

三、使用方法

Flink 读写 Doris 数据主要有两种方式：

·SQL

·DataStream

四、参数配置

Flink Doris Connector Sink 的内部实现是通过 Stream Load 服务向 Doris 写入数据, 同时也支持 Stream Load 请求参数的配置设置，具体参数可参考Stream Load手册，配置方法如下：

·SQL 使用 WITH 参数 sink.properties. 配置

·DataStream 使用方法 DorisExecutionOptions.builder().setStreamLoadProp(Properties) 配置

五、SQL

·Source（Doris 表作为数据源）

CREATE TABLE flink_doris_source ( name STRING, age INT, price DECIMAL(5,2), sale DOUBLE ) WITH ( 'connector' = 'doris', 'fenodes' = 'FE_IP:8030', 'table.identifier' = 'database.table', 'username' = 'root', 'password' = '*******');

·Sink（Doris 表作为导入目标表）

— enable checkpointSET 'execution.checkpointing.interval' = '10s';CREATE TABLE flink_doris_sink ( name STRING, age INT, price DECIMAL(5,2), sale DOUBLE ) WITH ( 'connector' = 'doris', 'fenodes' = 'FE_IP:8030', 'table.identifier' = 'db.table', 'username' = 'root', 'password' = '*******', 'sink.label-prefix' = 'doris_label');

·Insert

INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source

六、DataStream

·Source

DorisOptions.Builder builder = DorisOptions.builder() .setFenodes("FE_IP:8030") .setTableIdentifier("db.table") .setUsername("root") .setPassword("*******");DorisSource<List<?>> dorisSource = DorisSourceBuilder.<List<?>>builder() .setDorisOptions(builder.build()) .setDorisReadOptions(DorisReadOptions.builder().build()) .setDeserializer(new SimpleListDeserializationSchema()) .build();env.fromSource(dorisSource, WatermarkStrategy.noWatermarks(), "doris source").print();

·Sink
String 数据流

// enable checkpointenv.enableCheckpointing(10000);DorisSink.Builder<String> builder = DorisSink.builder();DorisOptions.Builder dorisBuilder = DorisOptions.builder();dorisBuilder.setFenodes("FE_IP:8030") .setTableIdentifier("db.table") .setUsername("root") .setPassword("*******");DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();executionBuilder.setLabelPrefix("label-doris"); //streamload label prefixbuilder.setDorisReadOptions(DorisReadOptions.builder().build()) .setDorisExecutionOptions(executionBuilder.build()) .setSerializer(new SimpleStringSerializer()) //serialize according to string .setDorisOptions(dorisBuilder.build());//mock string sourceList<Tuple2<String, Integer>> data = new ArrayList<>();data.add(new Tuple2<>("doris",1));DataStreamSource<Tuple2<String, Integer>> source = env.fromCollection(data);source.map((MapFunction<Tuple2<String, Integer>, String>) t -> t.f0 + "\\t" + t.f1) .sinkTo(builder.build());

RowData 数据流

// enable checkpointenv.enableCheckpointing(10000);//doris sink optionDorisSink.Builder<RowData> builder = DorisSink.builder();DorisOptions.Builder dorisBuilder = DorisOptions.builder();dorisBuilder.setFenodes("FE_IP:8030") .setTableIdentifier("db.table") .setUsername("root") .setPassword("*******");// json format to streamloadProperties properties = new Properties();properties.setProperty("format", "json");properties.setProperty("read_json_by_line", "true");DorisExecutionOptions.Builder executionBuilder = DorisExecutionOptions.builder();executionBuilder.setLabelPrefix("label-doris") //streamload label prefix .setStreamLoadProp(properties); //streamload params//flink rowdata‘s schemaString[] fields = {"city", "longitude", "latitude"};DataType[] types = {DataTypes.VARCHAR(256), DataTypes.DOUBLE(), DataTypes.DOUBLE()};builder.setDorisReadOptions(DorisReadOptions.builder().build()) .setDorisExecutionOptions(executionBuilder.build()) .setSerializer(RowDataSerializer.builder() //serialize according to rowdata .setFieldNames(fields) .setType("json") //json format .setFieldType(types).build()) .setDorisOptions(dorisBuilder.build());//mock rowdata sourceDataStream<RowData> source = env.fromElements("") .map(new MapFunction<String, RowData>() { @Override public RowData map(String value) throws Exception { GenericRowData genericRowData = new GenericRowData(3); genericRowData.setField(0, StringData.fromString("beijing")); genericRowData.setField(1, 116.405419); genericRowData.setField(2, 39.916927); return genericRowData; } });source.sinkTo(builder.build());

七、通用配置项

八、Doris 和 Flink 列类型映射关系

九、使用 Flink CDC 接入 Doris 示例（支持 Insert / Update / Delete 事件）

CREATE TABLE cdc_mysql_source ( id int ,name VARCHAR ,PRIMARY KEY (id) NOT ENFORCED) WITH ( 'connector' = 'mysql-cdc', 'hostname' = '127.**.**.**', 'port' = '3306', 'username' = 'root', 'password' = '******', 'database-name' = 'database', 'table-name' = 'table');– 支持删除事件同步(sink.enable-delete='true'),需要 Doris 表开启批量删除功能CREATE TABLE doris_sink (id INT,name STRING) WITH ( 'connector' = 'doris', 'fenodes' = '127.**.**.**:8030', 'table.identifier' = 'database.table', 'username' = 'root', 'password' = '******', 'sink.properties.format' = 'json', 'sink.properties.read_json_by_line' = 'true', 'sink.enable-delete' = 'true', 'sink.label-prefix' = 'doris_label');insert into doris_sink select id,name from cdc_mysql_source;

十、Java 示例

samples/doris-demo/fink-demo/ 下提供了 Java 版本的示例，可供参考，查看点击 这里。

十一、应用场景

使用 Flink Doris Connector 最适合的场景就是实时/批量同步源数据（从 Mysql，Oracle，PostgreSQL 等）到 Doris，使用 Flink 对 Doris 中的数据和其他数据源进行联合分析，也可以使用 Flink Doris Connector。

十二、其他

·Flink Doris Connector 主要是依赖 Checkpoint 进行流式写入，所以 Checkpoint 的间隔即为数据的可见延迟时间。

·为了保证 Flink 的 Exactly Once 语义，Flink Doris Connector 默认开启两阶段提交，Doris 在1.1版本后默认开启两阶段提交。1.0可通过修改 BE 参数开启，可参考 Stream load（本地文件）。

十三、常见问题

1. Bitmap 类型写入

CREATE TABLE bitmap_sink (dt int,page string,user_id int )WITH ( 'connector' = 'doris', 'fenodes' = '127.**.**.**:8030', 'table.identifier' = 'test.bitmap_test', 'username' = 'root', 'password' = '******', 'sink.label-prefix' = 'doris_label', 'sink.properties.columns' = 'dt,page,user_id,user_id=to_bitmap(user_id)')

2. errCode = 2, detailMessage = Label [label_0_1] has already been used, relate to txn [19650]
Exactly-Once 场景下，Flink Job 重启时必须从最新的 Checkpoint/Savepoint 启动，否则会报如上错误。
不要求 Exactly-Once 时，也可通过关闭2PC 提交（sink.enable-2pc=false） 或更换不同的 sink.label-prefix 解决。