Doris Connector

This article introduces of Doris Connector

Example

source:
   type: values
   name: ValuesSource

sink:
   type: doris
   name: Doris Sink
   fenodes: 127.0.0.1:8030
   username: root
   password: ""
   table.create.properties.replication_num: 1

pipeline:
   parallelism: 1

Connector Options

Option	Required	Default	Type	Description
type	required	(none)	String	Specify the Sink to use, here is 'doris'.
name	optional	(none)	String	Name of PipeLine
fenodes	required	(none)	String	Http address of Doris cluster FE, such as 127.0.0.1:8030
benodes	optional	(none)	String	Http address of Doris cluster BE, such as 127.0.0.1:8040
jdbc-url	optional	(none)	String	JDBC address of Doris cluster, for example: jdbc:mysql://127.0.0.1:9030/db
username	required	(none)	String	Username of Doris cluster
password	optional	(none)	String	Password for Doris cluster
auto-redirect	optional	false	Boolean	Whether to write through FE redirection and directly connect to BE to write
charset-encoding	optional	UTF-8	String	Charset encoding for doris http client, default UTF-8
sink.enable.batch-mode	optional	true	Boolean	Whether to use the batch method to write to Doris
sink.enable-delete	optional	true	Boolean	Whether to enable the delete function
sink.max-retries	optional	3	Integer	The max retry times if writing records to database failed.
sink.flush.queue-size	optional	2	Integer	Queue size for batch writing
sink.buffer-flush.max-rows	optional	50000	Integer	Maximum number of Flush records in a single batch
sink.buffer-flush.max-bytes	optional	10485760(10MB)	Integer	Maximum number of bytes flushed in a single batch
sink.buffer-flush.interval	optional	10s	String	Flush interval duration. If this time is exceeded, the data will be flushed asynchronously
sink.ignore.update-before	optional	true	Boolean	In the CDC scenario, when the primary key of the upstream is inconsistent with that of the downstream, the update-before data needs to be passed to the downstream as deleted data, otherwise the data cannot be deleted. The default is to ignore, that is, perform upsert semantics.
sink.enable-2pc	optional	false	Boolean	Enable 2PC while loading
sink.check-interval	optional	10000	Integer	Check exception with the interval (in milliseconds) while loading
sink.buffer-size	optional	1048576(1MB)	Integer	The buffer size (in bytes) to cache data for stream load
sink.buffer-count	optional	3	Integer	The buffer count to cache data for stream load
sink.label-prefix	optional	(none)	String	The unique label prefix used for stream load
sink.use-cache	optional	false	Boolean	Whether to use buffer cache for breakpoint resume
sink.properties.	optional	(none)	String	Parameters of StreamLoad. For example: sink.properties.strict_mode: true. See more about StreamLoad Properties
table.create.properties.*	optional	(none)	String	Create the Properties configuration of the table. For example: table.create.properties.replication_num: 1. See more about Doris Table Properties
table.create.auto-partition.properties.*	optional	(none)	String	Create the auto partition Properties configuration of the table. Currently, the partition function only supports date_trunc, the partition column only supports DATE or DATETIME types, and the Doris version must be later than 2.1.6. See more about Doris Auto Partitioning These properties are supported now: table.create.auto-partition.properties.includeA collection of tables after route to include, separated by commas, supports regular expressions; table.create.auto-partition.properties.excludeA collection of tables after route to exclude, separated by commas, supports regular expressions; table.create.auto-partition.properties.default-partition-keyThe default partition key; table.create.auto-partition.properties.default-partition-unitThe default partition unit; table.create.auto-partition.properties.DB.TABLE.partition-keyThe partition key of a specific table. If not set, the default partition key is used; table.create.auto-partition.properties.DB.TABLE.partition-unitThe partition unit of a specific table. If not set, the default partition unit is used. Note: 1: If the partition key is not DATE/DATETIME type, auto partition tables won't be created. 2: Doris AUTO RANGE PARTITION does not support NULLABLE columns as partition key, if Flink CDC get a NULL value or a NULLABLE partition key was added after the table was created, will automatically fill it with a default value(DATE:1970-01-01, DATETIME:1970-01-01 00:00:00), chose a suitable partition key is very important.

Data Type Mapping

Flink CDC Type	Doris Type	Note
TINYINT	TINYINT
SMALLINT	SMALLINT
INT	INT
BIGINT	BIGINT
DECIMAL	DECIMAL
FLOAT	FLOAT
DOUBLE	DOUBLE
BOOLEAN	BOOLEAN
DATE	DATE
TIMESTAMP [(p)]	DATETIME [(p)]
TIMESTAMP_LTZ [(p)]	DATETIME [(p)]
CHAR(n)	CHAR(n*3)	In Doris, strings are stored in UTF-8 encoding, so English characters occupy 1 byte and Chinese characters occupy 3 bytes. The length here is multiplied by 3. The maximum length of CHAR is 255. Once exceeded, it will automatically be converted to VARCHAR type.
VARCHAR(n)	VARCHAR(n*3)	Same as above. The length here is multiplied by 3. The maximum length of VARCHAR is 65533. Once exceeded, it will automatically be converted to STRING type.
BINARY(n)	STRING
VARBINARY(N)	STRING
TIME	STRING	Doris does not support the TIME data type, it needs to be converted to STRING type for compatibility.
STRING	STRING