Flink Doris Connector

    • The Flink Doris Connector can support operations (read, insert, modify, delete) data stored in Doris through Flink.
    • table can be mapped to DataStream or Table.

    Build and Install

    Execute following command in source dir:

    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.

    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

    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

    • SQL
    • DataStream
    • DataSet

    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)
    • 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
    • Insert
    1. INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source
    • 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();

    Json Stream

    1. Properties pro = new Properties();
    2. pro.setProperty("format", "json");
    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

    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.     ));
    • 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.setField(0, StringData.fromString("北京"));
    7.             genericRowData.setField(2, 39.916927);
    8.             return genericRowData;
    9.         }
    10.     });
    12. DorisOptions dorisOptions = DorisOptions.builder()
    13.     .setFenodes("FE_IP:8030")
    14.     .setTableIdentifier("db.table")
    15.     .setUsername("root")
    16.     .setPassword("").build();
    17. DorisReadOptions readOptions = DorisReadOptions.defaults();
    18. DorisExecutionOptions executionOptions = DorisExecutionOptions.defaults();
    20. LogicalType[] types = {new VarCharType(), new DoubleType(), new DoubleType()};
    21. String[] fields = {"city", "longitude", "latitude"};
    23. DorisDynamicOutputFormat outputFormat = new DorisDynamicOutputFormat(
    24.     dorisOptions, readOptions, executionOptions, types, fields
    25.     );
    27. outputFormat.open(0, 1);
    28. data.output(outputFormat);
    29. outputFormat.close();

    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;