Skip to main content
Version: 1.7.0

Doris

Overview

  • Doris Load node supports writing data to the Doris database.
  • Two modes are supported for sink to Doris: Single-sink for specify fixed database name and table name to sink. Multi-sink for custom database name and table name according to src format, which suitable for scenarios such as multi-table writing or whole database synchronization.
  • This document describes how to set up a Doris Load node to sink to Doris.

Supported Version

Load NodeDoris version
Doris0.13+

Dependencies

In order to set up the Doris Load node, the dependency information needed to use a build automation tool such as Maven or SBT is provided below.

Maven dependency

<dependency>
<groupId>org.apache.inlong</groupId>
<artifactId>sort-connector-doris</artifactId>
<version>1.7.0</version>
</dependency>

Prepare

Create MySql Extract table

  • For Single-sink: Create a table cdc.cdc_mysql_source in the MySQL database. The command is as follows:
[root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
mysql> use cdc;
Database changed
mysql> CREATE TABLE `cdc_mysql_source` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(64) DEFAULT NULL,
`dr` tinyint(3) DEFAULT 0,
PRIMARY KEY (`id`)
);
Query OK, 0 rows affected (0.02 sec)

mysql> insert into cdc_mysql_source values(1, 'zhangsan', 0),(2, 'lisi', 0),(3, 'wangwu', 0);
Query OK, 3 rows affected (0.01 sec)
Records: 3 Duplicates: 0 Warnings: 0

mysql> select * from cdc_mysql_source;
+----+----------+----+
| id | name | dr |
+----+----------+----+
| 1 | zhangsan | 0 |
| 2 | lisi | 0 |
| 3 | wangwu | 0 |
+----+----------+----+
3 rows in set (0.07 sec)
  • For Multi-sink: Create tables user_db.user_id_nameuser_db.user_id_score in the MySQL database. The command is as follows:
[root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
mysql> use user_db;
Database changed
mysql> CREATE TABLE `user_id_name` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`name` varchar(64) DEFAULT NULL
PRIMARY KEY (`id`)
);
Query OK, 0 rows affected (0.02 sec)

mysql> CREATE TABLE `user_id_score` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`score` double default 0,
PRIMARY KEY (`id`)
);
Query OK, 0 rows affected (0.02 sec)

mysql> insert into user_id_name values(1001, 'lily'),(1002, 'tom'),(1003, 'alan');
Query OK, 3 rows affected (0.01 sec)
Records: 3 Duplicates: 0 Warnings: 0

mysql> insert into user_id_score values(1001, 99),(1002, 96),(1003, 98);
Query OK, 3 rows affected (0.01 sec)
Records: 3 Duplicates: 0 Warnings: 0

mysql> select * from user_id_name;
+------+--------+
| id | name |
+------+--------+
| 1001 | lily |
| 1002 | tom |
| 1003 | alan |
+----+----------+
3 rows in set (0.07 sec)

mysql> select * from user_id_score;
+------+------+
| id | name |
+------+------+
| 1001 | 99 |
| 1002 | 96 |
| 1003 | 98 |
+----+--------+
3 rows in set (0.07 sec)

Create Doris Load table

  • For Single-sink: Create a table cdc.cdc_doris_sink in the Doris database. The command is as follows:
[root@fe001 ~]# mysql -u root -h localhost -P 9030 -p000000
mysql> use cdc;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changed

mysql> CREATE TABLE `cdc_doris_sink` (
`id` int(11) NOT NULL COMMENT "user id",
`name` varchar(50) NOT NULL COMMENT "user name",
`dr` tinyint(4) NULL COMMENT "delete tag"
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);
Query OK, 0 rows affected (0.06 sec)
  • For Multi-sink: Create tables user_db.doris_user_id_nameuser_db.doris_user_id_score in the Doris database. The command is as follows:
[root@fe001 ~]# mysql -u root -h localhost -P 9030 -p000000
mysql> use user_db;
Reading table information for completion of table and column names
You can turn off this feature to get a quicker startup with -A
Database changed

mysql> CREATE TABLE `doris_user_id_name` (
`id` int(11) NOT NULL COMMENT "用户id",
`name` varchar(50) NOT NULL COMMENT "昵称"
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);
Query OK, 0 rows affected (0.06 sec)

mysql> CREATE TABLE `doris_user_id_score` (
`id` int(11) NOT NULL COMMENT "用户id",
`score` double default 0
) ENGINE=OLAP
UNIQUE KEY(`id`)
COMMENT "OLAP"
DISTRIBUTED BY HASH(`id`) BUCKETS 1
PROPERTIES (
"replication_allocation" = "tag.location.default: 1"
);
Query OK, 0 rows affected (0.06 sec)

How to create a Doris Load Node

Usage for SQL API

  • For Single-sink: Doris load
[root@tasknode001 flink-1.13.5]# ./bin/sql-client.sh -l ./opt/connectors/mysql-cdc-inlong/ -l ./opt/connectors/doris/
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
[INFO] Session property has been set.

Flink SQL> SET 'table.dynamic-table-options.enabled' = 'true';
[INFO] Session property has been set.

Flink SQL> CREATE TABLE cdc_mysql_source (
> id int
> ,name VARCHAR
> ,dr TINYINT
> ,PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'mysql-cdc-inlong',
> 'hostname' = 'localhost',
> 'port' = '3306',
> 'username' = 'root',
> 'password' = '123456',
> 'database-name' = 'cdc',
> 'table-name' = 'cdc_mysql_source'
> );
[INFO] Execute statement succeed.

Flink SQL> CREATE TABLE cdc_doris_sink (
> id INT,
> name STRING,
> dr TINYINT
> ) WITH (
> 'connector' = 'doris-inlong',
> 'fenodes' = 'localhost:8030',
> 'table.identifier' = 'cdc.cdc_doris_sink',
> 'username' = 'root',
> 'password' = '000000',
> 'sink.properties.format' = 'json',
> 'sink.properties.strip_outer_array' = 'true',
> 'sink.enable-delete' = 'true'
> );
[INFO] Execute statement succeed.

-- Support delete event synchronization (sink.enable-delete='true'), requires Doris table to enable batch delete function
Flink SQL> insert into cdc_doris_sink select * from cdc_mysql_source /*+ OPTIONS('server-id'='5402') */;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 5f89691571d7b3f3ca446589e3d0c3d3
  • For Single-sink: Doris load
./bin/sql-client.sh -l ./opt/connectors/mysql-cdc-inlong/ -l ./opt/connectors/doris/
Flink SQL> SET 'execution.checkpointing.interval' = '3s';
[INFO] Session property has been set.

Flink SQL> SET 'table.dynamic-table-options.enabled' = 'true';
[INFO] Session property has been set.

Flink SQL> CREATE TABLE cdc_mysql_source (
> id int
> ,name VARCHAR
> ,dr TINYINT
> ,PRIMARY KEY (id) NOT ENFORCED
> ) WITH (
> 'connector' = 'mysql-cdc-inlong',
> 'hostname' = 'localhost',
> 'port' = '3306',
> 'username' = 'root',
> 'password' = '123456',
> 'database-name' = 'test',
> 'table-name' = 'cdc_mysql_source'
> );
[INFO] Execute statement succeed.

Flink SQL> CREATE TABLE cdc_doris_sink (
> id INT,
> name STRING,
> dr TINYINT
> ) WITH (
> 'connector' = 'doris-inlong',
> 'fenodes' = 'localhost:8030',
> 'username' = 'root',
> 'password' = '000000',
> 'sink.enable-delete' = 'true',
> 'sink.multiple.enable' = 'true',
> 'sink.multiple.format' = 'canal-json',
> 'sink.multiple.database-pattern' = '${database}',
> 'sink.multiple.table-pattern' = 'doris_${table}'
> );
[INFO] Execute statement succeed.

-- Support delete event synchronization (sink.enable-delete='true'), requires Doris table to enable batch delete function
Flink SQL> insert into cdc_doris_sink select * from cdc_mysql_source /*+ OPTIONS('server-id'='5402') */;
[INFO] Submitting SQL update statement to the cluster...
[INFO] SQL update statement has been successfully submitted to the cluster:
Job ID: 30feaa0ede92h6b6e25ea0cfda26df5e

Usage for InLong Dashboard

TODO: It will be supported in the future.

Usage for InLong Manager Client

TODO: It will be supported in the future.

Doris Load Node Options

OptionRequiredDefaultTypeDescription
connectorrequired(none)stringSpecify which connector to use, valid values are: doris
fenodesrequired(none)stringDoris FE http address, support multiple addresses, separated by commas
table.identifierrequired(none)stringDoris table identifier, eg, db1.tbl1
usernamerequired(none)stringDoris username
passwordrequired(none)stringDoris password
doris.request.retriesoptional3intNumber of retries to send requests to Doris
doris.request.connect.timeout.msoptional30000intConnection timeout for sending requests to Doris
doris.request.read.timeout.msoptional30000intRead timeout for sending request to Doris
doris.request.query.timeout.soptional3600intQuery the timeout time of Doris, the default is 1 hour, -1 means no timeout limit
doris.request.tablet.sizeoptionalInteger.MAX_VALUEintThe number of Doris Tablets corresponding to an Partition. The smaller this value is set, the more partitions will be generated. This will increase the parallelism on the flink side, but at the same time will cause greater pressure on Doris.
doris.batch.sizeoptional1024intThe maximum number of rows to read data from BE at one time. Increasing this value can reduce the number of connections between Flink and Doris. Thereby reducing the extra time overhead caused by network delay.
doris.exec.mem.limitoptional2147483648longMemory limit for a single query. The default is 2GB, in bytes.
doris.deserialize.arrow.asyncoptionalfalsebooleanWhether to support asynchronous conversion of Arrow format to RowBatch required for flink-doris-connector iteration
doris.deserialize.queue.sizeoptional64intAsynchronous conversion of the internal processing queue in Arrow format takes effect when doris.deserialize.arrow.async is true
doris.read.fieldoptional(none)stringList of column names in the Doris table, separated by commas
doris.filter.queryoptional(none)stringFilter expression of the query, which is transparently transmitted to Doris. Doris uses this expression to complete source-side data filtering.
sink.batch.sizeoptional10000intMaximum number of lines in a single write BE
sink.max-retriesoptional1intNumber of retries after writing BE failed
sink.batch.intervaloptional10sstringThe flush interval, after which the asynchronous thread will write the data in the cache to BE. The default value is 10 second, and the time units are ms, s, min, h, and d. Set to 0 to turn off periodic writing.
sink.properties.*optional(none)stringThe stream load parameters.

eg:
sink.properties.column_separator' = ','

Setting 'sink.properties.escape_delimiters' = 'true' if you want to use a control char as a separator, so that such as '\x01' will translate to binary 0x01

Support JSON format import, you need to enable both 'sink.properties.format' ='json' and 'sink.properties.strip_outer_array' ='true'

Support CSV format data writing if 'sink.properties.format'='csv' is setted.
sink.enable-deleteoptionaltruebooleanWhether to enable deletion. This option requires Doris table to enable batch delete function (0.15+ version is enabled by default), and only supports Unique model.
sink.multiple.enableoptionalfalsebooleanDetermine whether to support multiple sink writing, default is false. when sink.multiple.enable is true, need sink.multiple.formatsink.multiple.database-patternsink.multiple.table-pattern be correctly set.
sink.multiple.formatoptional(none)stringThe format of multiple sink, it represents the real format of the raw binary data. can be canal-json or debezium-json at present. See kafka -- Dynamic Topic Extraction for more details.
sink.multiple.database-patternoptional(none)stringExtract database name from the raw binary data, this is only used in the multiple sink writing scenario.
sink.multiple.table-patternoptional(none)stringExtract table name from the raw binary data, this is only used in the multiple sink writing scenario.
sink.multiple.ignore-single-table-errorsoptionaltruebooleanWhether ignore the single table erros when multiple sink writing scenario. When it is true,sink continue when one table occur exception, only stop the exception table sink. When it is false, stop the whole sink when one table occur exception.
inlong.metric.labelsoptional(none)stringInlong metric label, format of value is groupId={groupId}&streamId={streamId}&nodeId={nodeId}.
sink.multiple.schema-update.policyoptional(none)stringIf sink data fields do not match doris table, such as table does not exsit or string data is over length, doris server will throw an exception.

When this option is THROW_WITH_STOP, the exception will be thrown up to flink framework, flink will restart task automatically, trying to resume the task.

When this option is STOP_PARTIAL, doris connector will stop writing into this table, other tables are written normally. The exception will be logging but not thrown up.

When this option is LOG_WITH_IGNORE, doris connector only log the error, not throw up. Doris connector will try to write to doris server again when receiving new source data.
dirty.ignoreoptinal(none)booleanWhen writing data into doris table, errors may be thrown by doris server as table does not exist or data is over length.

When this option is true, and dirty.side-output.* properties are configed correctly, dirty data can be written to Amazon S3 or Tencent Colud COS storage. Dirty data metrics will also be collected automatically.

When this option is false, only dirty data metrics will be collected, but dirty data will not be archived.
dirty.side-output.enableoptinal(none)booleanWhen this option is ture and other options about S3 or COS is configed correctly, dirty data archiving will works. When false, dirty data archiving will not work.
dirty.side-output.connectoroptinal(none)strings3 or log are supported now.

When log, doris connector only log the dirty data, not archive data.

When s3, doris connector can write dirty data to S3 or COS.
dirty.side-output.s3.bucketoptinal(none)stringThe bucket name of S3 or COS
dirty.side-output.s3.endpointoptinal(none)stringThe endpoint of S3 or COS
dirty.side-output.s3.keyoptinal(none)stringThe key of S3 or COS
dirty.side-output.s3.regionoptinal(none)stringThe region of S3 or COS
dirty.side-output.line-delimiteroptinal(none)stringThe line delimiter of dirty data
dirty.side-output.field-delimiteroptinal(none)stringThe field delimiter of dirty data
dirty.side-output.s3.secret-key-idoptinal(none)stringThe secret key of S3 or COS
dirty.side-output.s3.access-key-idoptinal(none)stringThe access key of S3 or COS
dirty.side-output.formatoptinal(none)stringThe format of dirty data archiving, supports json or csv
dirty.side-output.log-tagoptinal(none)stringThe log tag of dirty data. Doris connector uses lags to distinguish which doris database and table the dirty data will be written to.
dirty.identifieroptinal(none)stringThe file name of drity data which written to S3 or COS.
dirty.side-output.labelsoptinal(none)stringEvery dirty data line contains label and business data fields. Label is in front, and business data is at end.

Data Type Mapping

Doris TypeFlink Type
NULL_TYPENULL
BOOLEANBOOLEAN
TINYINTTINYINT
SMALLINTSMALLINT
INTINT
BIGINTBIGINT
FLOATFLOAT
DOUBLEDOUBLE
DATESTRING
DATETIMESTRING
DECIMALDECIMAL
CHARSTRING
LARGEINTSTRING
VARCHARSTRING
DECIMALV2DECIMAL
TIMEDOUBLE
HLLUnsupported datatype

See flink-doris-connector for more details.