Flink Doris Connector

Flink Doris Connector 可以支持通过 Flink 操作（读取、插入、修改、删除） Doris 中存储的数据。

代码库地址：<https://github.com/apache/incubator-doris-flink-connector>

• 可以将Doris表映射为DataStream或者Table。

注意：

1. 修改和删除只支持在Unique Key模型上
2. 目前的删除是支持Flink CDC的方式接入数据实现自动删除，如果是其他数据接入的方式删除需要自己实现。Flink CDC的数据删除使用方式参照本文档最后一节

版本兼容

编译与安装

在源码目录下执行：

1. sh build.sh --flink 1.11.6 --scala 2.12 # flink 1.11.6 scala 2.12

注：如果你是从 tag 检出的源码，则可以直接执行 sh build.sh --tag，而无需指定 flink 和 scala 的版本。因为 tag 源码中的版本是固定的。比如 1.13.5-2.12-1.0.1 表示 flink 版本 1.13.5，scala 版本 2.12，connector 版本 1.0.1。

编译成功后，会在 output/ 目录下生成文件，如：doris-flink-1.13.5-2.12-1.0.1-SNAPSHOT.jar 。将此文件复制到 Flink 的 ClassPath 中即可使用 Flink-Doris-Connector 。例如， Local 模式运行的 Flink ，将此文件放入 jars/ 文件夹下。 Yarn 集群模式运行的 Flink ，则将此文件放入预部署包中。

备注

1. doris FE 要在配置中配置启用http v2
2. Scala版本目前只支持2.12.x版本

conf/fe.conf

1. enable_http_server_v2 = true

使用Maven 管理

添加 maven 依赖

1. <dependency>
2. <groupId>org.apache.doris</groupId>
3. <artifactId>doris-flink-connector</artifactId>
4. <version>1.11.6-2.12-SNAPSHOT</version>
5. </dependency>

备注

1.11.6 可以根据flink 版本替换成替换成 1.12.7 或者 1.13.5

使用方法

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

• SQL
• DataStream
• DataSet

参数配置

Flink Doris Connector Sink的内部实现是通过 Stream load 服务向Doris写入数据, 同时也支持 Stream load 请求参数的配置设定

参数配置方法

• SQL 使用WITH参数sink.properties.配置
• DataStream 使用方法DorisExecutionOptions.builder().setStreamLoadProp(Properties)配置

SQL

• Source

1. CREATE TABLE flink_doris_source (
2. name STRING,
3. age INT,
4. price DECIMAL(5,2),
5. sale DOUBLE
6. )
7. WITH (
8. 'connector' = 'doris',
9. 'fenodes' = '$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT',
10. 'table.identifier' = '$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME',
11. 'username' = '$YOUR_DORIS_USERNAME',
12. 'password' = '$YOUR_DORIS_PASSWORD'
13. );

• Sink

1. CREATE TABLE flink_doris_sink (
2. name STRING,
3. age INT,
4. price DECIMAL(5,2),
5. sale DOUBLE
6. )
7. WITH (
8. 'connector' = 'doris',
9. 'fenodes' = '$YOUR_DORIS_FE_HOSTNAME:$YOUR_DORIS_FE_RESFUL_PORT',
10. 'table.identifier' = '$YOUR_DORIS_DATABASE_NAME.$YOUR_DORIS_TABLE_NAME',
11. 'username' = '$YOUR_DORIS_USERNAME',
12. 'password' = '$YOUR_DORIS_PASSWORD'
13. );

• Insert

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

DataStream

• Source

1. Properties properties = new Properties();
2. properties.put("fenodes","FE_IP:8030");
3. properties.put("username","root");
4. properties.put("password","");
5. properties.put("table.identifier","db.table");
6. env.addSource(new DorisSourceFunction(
7. new DorisStreamOptions(properties),
8. new SimpleListDeserializationSchema()
9. )
10. ).print();

• Sink

Json数据流

1. Properties pro = new Properties();
2. pro.setProperty("format", "json");
3. pro.setProperty("strip_outer_array", "true");
4. env.fromElements(
5. "{\"longitude\": \"116.405419\", \"city\": \"北京\", \"latitude\": \"39.916927\"}"
6. )
7. .addSink(
8. DorisSink.sink(
9. DorisReadOptions.builder().build(),
10. DorisExecutionOptions.builder()
11. .setBatchSize(3)
12. .setBatchIntervalMs(0l)
13. .setMaxRetries(3)
14. .setStreamLoadProp(pro).build(),
15. DorisOptions.builder()
16. .setFenodes("FE_IP:8030")
17. .setTableIdentifier("db.table")
18. .setUsername("root")
19. .setPassword("").build()
20. ));

Json数据流

1. env.fromElements(
2. "{\"longitude\": \"116.405419\", \"city\": \"北京\", \"latitude\": \"39.916927\"}"
3. )
4. .addSink(
5. DorisSink.sink(
6. DorisOptions.builder()
7. .setFenodes("FE_IP:8030")
8. .setTableIdentifier("db.table")
9. .setUsername("root")
10. .setPassword("").build()
11. ));

RowData数据流

1. DataStream<RowData> source = env.fromElements("")
2. .map(new MapFunction<String, RowData>() {
3. @Override
4. public RowData map(String value) throws Exception {
5. GenericRowData genericRowData = new GenericRowData(3);
6. genericRowData.setField(0, StringData.fromString("北京"));
7. genericRowData.setField(1, 116.405419);
8. genericRowData.setField(2, 39.916927);
9. return genericRowData;
10. }
11. });

13. String[] fields = {"city", "longitude", "latitude"};
14. LogicalType[] types = {new VarCharType(), new DoubleType(), new DoubleType()};

16. source.addSink(
17. DorisSink.sink(
18. fields,
19. types,
20. DorisReadOptions.builder().build(),
21. DorisExecutionOptions.builder()
22. .setBatchSize(3)
23. .setBatchIntervalMs(0L)
24. .setMaxRetries(3)
25. .build(),
26. DorisOptions.builder()
27. .setFenodes("FE_IP:8030")
28. .setTableIdentifier("db.table")
29. .setUsername("root")
30. .setPassword("").build()
31. ));

DataSet

• Sink

1. MapOperator<String, RowData> data = env.fromElements("")
2. .map(new MapFunction<String, RowData>() {
3. @Override
4. public RowData map(String value) throws Exception {
5. GenericRowData genericRowData = new GenericRowData(3);
6. genericRowData.setField(0, StringData.fromString("北京"));
7. genericRowData.setField(1, 116.405419);
8. genericRowData.setField(2, 39.916927);
9. return genericRowData;
10. }
11. });

13. DorisOptions dorisOptions = DorisOptions.builder()
14. .setFenodes("FE_IP:8030")
15. .setTableIdentifier("db.table")
16. .setUsername("root")
17. .setPassword("").build();
18. DorisReadOptions readOptions = DorisReadOptions.defaults();
19. DorisExecutionOptions executionOptions = DorisExecutionOptions.defaults();

21. LogicalType[] types = {new VarCharType(), new DoubleType(), new DoubleType()};
22. String[] fields = {"city", "longitude", "latitude"};

24. DorisDynamicOutputFormat outputFormat = new DorisDynamicOutputFormat(
25. dorisOptions, readOptions, executionOptions, types, fields
26. );

28. outputFormat.open(0, 1);
29. data.output(outputFormat);
30. outputFormat.close();

配置

通用配置项

Doris 和 Flink 列类型映射关系

使用Flink CDC接入Doris示例（支持insert/update/delete事件）

1. CREATE TABLE cdc_mysql_source (
2. id int
3. ,name VARCHAR
4. ,PRIMARY KEY (id) NOT ENFORCED
5. ) WITH (
6. 'connector' = 'mysql-cdc',
7. 'hostname' = '127.0.0.1',
8. 'port' = '3306',
9. 'username' = 'root',
10. 'password' = 'password',
11. 'database-name' = 'database',
12. 'table-name' = 'table'
13. );

15. -- 支持删除事件同步(sink.enable-delete='true'),需要Doris表开启批量删除功能
16. CREATE TABLE doris_sink (
17. id INT,
18. name STRING
19. )
20. WITH (
21. 'connector' = 'doris',
22. 'fenodes' = '127.0.0.1:8030',
23. 'table.identifier' = 'database.table',
24. 'username' = 'root',
25. 'password' = '',
26. 'sink.properties.format' = 'json',
27. 'sink.properties.strip_outer_array' = 'true',
28. 'sink.enable-delete' = 'true'
29. );

31. insert into doris_sink select id,name from cdc_mysql_source;