Flink Doris Connector

• The Flink Doris Connector can support operations (read, insert, modify, delete) data stored in Doris through Flink.

Github: <https://github.com/apache/incubator-doris-flink-connector>

• Doris table can be mapped to DataStream or Table.

Note:

1. Modification and deletion are only supported on the Unique Key model
2. The current deletion is to support Flink CDC to access data to achieve automatic deletion. If it is to delete other data access methods, you need to implement it yourself. For the data deletion usage of Flink CDC, please refer to the last section of this document

Version Compatibility

Build and Install

Execute following command in source dir:

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

Note: If you check out the source code from tag, you can just run sh build.sh --tag without specifying the flink and scala versions. This is because the version in the tag source code is fixed. For example, 1.13.5-2.12-1.0.1 means flink version 1.13.5, scala version 2.12, and connector version 1.0.1.

After successful compilation, the file doris-flink-1.13.5-2.12-1.0.1-SNAPSHOT.jar will be generated in the output/ directory. Copy this file to ClassPath in Flink to use Flink-Doris-Connector. For example, Flink running in Local mode, put this file in the jars/ folder. Flink running in Yarn cluster mode, put this file in the pre-deployment package.

Remarks:

1. Doris FE should be configured to enable http v2 in the configuration
2. Scala version currently only supports 2.12.x version

conf/fe.conf

1. enable_http_server_v2 = true

Using Maven

Add Dependency

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>

Remarks

1.11.6 can be substitute with 1.12.7 or 1.13.5 base on flink version you are using

How to use

There are three ways to use Flink Doris Connector.

• SQL
• DataStream
• DataSet

Parameters Configuration

Flink Doris Connector Sink writes data to Doris by the Stream load, and also supports the configurations of Stream load

• SQL configured by sink.properties. in the WITH
• DataStream configured by 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 Stream

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 Stream

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 Stream

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();

General

Doris & Flink Column Type Mapping

An example of using Flink CDC to access Doris (supports insert/update/delete events)

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. -- Support delete event synchronization (sink.enable-delete='true'), requires Doris table to enable batch delete function
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;