Doris

概览

Doris Load 节点支持将数据写入 Doris 数据库。 支持单表写入和多表写入两种模式:单表写入为指定固定库名表名写入;多表写入支持根据源端数据格式自定义库名表名写入,适用于源端多表写入或者整库同步等场景。 本文档介绍如何设置 Doris Load 节点实现写入 Doris 数据库表。

支持的版本

Load 节点Doris 版本
Doris0.13+

依赖

为了设置 Doris Load 节点, 下面提供了使用构建自动化工具(例如 Maven 或 SBT)所需要的依赖信息。

Maven 依赖

  1. <dependency>
  2. <groupId>org.apache.inlong</groupId>
  3. <artifactId>sort-connector-doris</artifactId>
  4. <version>1.7.0</version>
  5. </dependency>

准备

创建 MySQL Extract 表

  • 单表写入:在 MySQL cdc 数据库中创建表 cdc_mysql_source。 命令如下:
  1. [root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
  2. mysql> use cdc;
  3. Database changed
  4. mysql> CREATE TABLE `cdc_mysql_source` (
  5. `id` int(11) NOT NULL AUTO_INCREMENT,
  6. `name` varchar(64) DEFAULT NULL,
  7. `dr` tinyint(3) DEFAULT 0,
  8. PRIMARY KEY (`id`)
  9. );
  10. Query OK, 0 rows affected (0.02 sec)
  11. mysql> insert into cdc_mysql_source values(1, 'zhangsan', 0),(2, 'lisi', 0),(3, 'wangwu', 0);
  12. Query OK, 3 rows affected (0.01 sec)
  13. Records: 3 Duplicates: 0 Warnings: 0
  14. mysql> select * from cdc_mysql_source;
  15. +----+----------+----+
  16. | id | name | dr |
  17. +----+----------+----+
  18. | 1 | zhangsan | 0 |
  19. | 2 | lisi | 0 |
  20. | 3 | wangwu | 0 |
  21. +----+----------+----+
  22. 3 rows in set (0.07 sec)
  • 多表写入:在 MySQL user_db 数据库中创建表 user_id_nameuser_id_score。 命令如下:
  1. [root@fe001 ~]# mysql -u root -h localhost -P 3306 -p123456
  2. mysql> use user_db;
  3. Database changed
  4. mysql> CREATE TABLE `user_id_name` (
  5. `id` int(11) NOT NULL AUTO_INCREMENT,
  6. `name` varchar(64) DEFAULT NULL
  7. PRIMARY KEY (`id`)
  8. );
  9. Query OK, 0 rows affected (0.02 sec)
  10. mysql> CREATE TABLE `user_id_score` (
  11. `id` int(11) NOT NULL AUTO_INCREMENT,
  12. `score` double default 0,
  13. PRIMARY KEY (`id`)
  14. );
  15. Query OK, 0 rows affected (0.02 sec)
  16. mysql> insert into user_id_name values(1001, 'lily'),(1002, 'tom'),(1003, 'alan');
  17. Query OK, 3 rows affected (0.01 sec)
  18. Records: 3 Duplicates: 0 Warnings: 0
  19. mysql> insert into user_id_score values(1001, 99),(1002, 96),(1003, 98);
  20. Query OK, 3 rows affected (0.01 sec)
  21. Records: 3 Duplicates: 0 Warnings: 0
  22. mysql> select * from user_id_name;
  23. +------+--------+
  24. | id | name |
  25. +------+--------+
  26. | 1001 | lily |
  27. | 1002 | tom |
  28. | 1003 | alan |
  29. +----+----------+
  30. 3 rows in set (0.07 sec)
  31. mysql> select * from user_id_score;
  32. +------+------+
  33. | id | name |
  34. +------+------+
  35. | 1001 | 99 |
  36. | 1002 | 96 |
  37. | 1003 | 98 |
  38. +----+--------+
  39. 3 rows in set (0.07 sec)

创建 Doris Load 表

  • 单表写入:在 Doris cdc数据库中创建表cdc_doris_sink。命令如下:
  1. [root@fe001 ~]# mysql -u root -h localhost -P 9030 -p000000
  2. mysql> use cdc;
  3. Reading table information for completion of table and column names
  4. You can turn off this feature to get a quicker startup with -A
  5. Database changed
  6. mysql> CREATE TABLE `cdc_doris_sink` (
  7. `id` int(11) NOT NULL COMMENT "用户id",
  8. `name` varchar(50) NOT NULL COMMENT "昵称",
  9. `dr` tinyint(4) NULL COMMENT "逻辑删除"
  10. ) ENGINE=OLAP
  11. UNIQUE KEY(`id`)
  12. COMMENT "OLAP"
  13. DISTRIBUTED BY HASH(`id`) BUCKETS 1
  14. PROPERTIES (
  15. "replication_allocation" = "tag.location.default: 1"
  16. );
  17. Query OK, 0 rows affected (0.06 sec)
  • 多表写入:在 Doris user_db数据库中创建表doris_user_id_namedoris_user_id_score。命令如下:
  1. [root@fe001 ~]# mysql -u root -h localhost -P 9030 -p000000
  2. mysql> use user_db;
  3. Reading table information for completion of table and column names
  4. You can turn off this feature to get a quicker startup with -A
  5. Database changed
  6. mysql> CREATE TABLE `doris_user_id_name` (
  7. `id` int(11) NOT NULL COMMENT "用户id",
  8. `name` varchar(50) NOT NULL COMMENT "昵称"
  9. ) ENGINE=OLAP
  10. UNIQUE KEY(`id`)
  11. COMMENT "OLAP"
  12. DISTRIBUTED BY HASH(`id`) BUCKETS 1
  13. PROPERTIES (
  14. "replication_allocation" = "tag.location.default: 1"
  15. );
  16. Query OK, 0 rows affected (0.06 sec)
  17. mysql> CREATE TABLE `doris_user_id_score` (
  18. `id` int(11) NOT NULL COMMENT "用户id",
  19. `score` double default 0
  20. ) ENGINE=OLAP
  21. UNIQUE KEY(`id`)
  22. COMMENT "OLAP"
  23. DISTRIBUTED BY HASH(`id`) BUCKETS 1
  24. PROPERTIES (
  25. "replication_allocation" = "tag.location.default: 1"
  26. );
  27. Query OK, 0 rows affected (0.06 sec)

如何创建 Doris Load 节点

SQL API 用法

  • 单表写入: Doris 单表写入
  1. [root@tasknode001 flink-1.13.5]# ./bin/sql-client.sh -l ./opt/connectors/mysql-cdc-inlong/ -l ./opt/connectors/doris/
  2. Flink SQL> SET 'execution.checkpointing.interval' = '3s';
  3. [INFO] Session property has been set.
  4. Flink SQL> SET 'table.dynamic-table-options.enabled' = 'true';
  5. [INFO] Session property has been set.
  6. Flink SQL> CREATE TABLE cdc_mysql_source (
  7. > id int
  8. > ,name VARCHAR
  9. > ,dr TINYINT
  10. > ,PRIMARY KEY (id) NOT ENFORCED
  11. > ) WITH (
  12. > 'connector' = 'mysql-cdc-inlong',
  13. > 'hostname' = 'localhost',
  14. > 'port' = '3306',
  15. > 'username' = 'root',
  16. > 'password' = '123456',
  17. > 'database-name' = 'cdc',
  18. > 'table-name' = 'cdc_mysql_source'
  19. > );
  20. [INFO] Execute statement succeed.
  21. Flink SQL> CREATE TABLE cdc_doris_sink (
  22. > id INT,
  23. > name STRING,
  24. > dr TINYINT
  25. > ) WITH (
  26. > 'connector' = 'doris-inlong',
  27. > 'fenodes' = 'localhost:8030',
  28. > 'table.identifier' = 'cdc.cdc_doris_sink',
  29. > 'username' = 'root',
  30. > 'password' = '000000',
  31. > 'sink.properties.format' = 'json',
  32. > 'sink.properties.strip_outer_array' = 'true',
  33. > );
  34. [INFO] Execute statement succeed.
  35. -- 支持删除事件同步(sink.enable-delete='true'), 需要 Doris 表开启批量删除功能
  36. Flink SQL> insert into cdc_doris_sink select * from cdc_mysql_source /*+ OPTIONS('server-id'='5402') */;
  37. [INFO] Submitting SQL update statement to the cluster...
  38. [INFO] SQL update statement has been successfully submitted to the cluster:
  39. Job ID: 5f89691571d7b3f3ca446589e3d0c3d3
  • 多表写入: Doris 多表写入
  1. ./bin/sql-client.sh -l ./opt/connectors/mysql-cdc-inlong/ -l ./opt/connectors/doris/
  2. Flink SQL> SET 'execution.checkpointing.interval' = '3s';
  3. [INFO] Session property has been set.
  4. Flink SQL> SET 'table.dynamic-table-options.enabled' = 'true';
  5. [INFO] Session property has been set.
  6. Flink SQL> CREATE TABLE cdc_mysql_source (
  7. > id int
  8. > ,name VARCHAR
  9. > ,dr TINYINT
  10. > ,PRIMARY KEY (id) NOT ENFORCED
  11. > ) WITH (
  12. > 'connector' = 'mysql-cdc-inlong',
  13. > 'hostname' = 'localhost',
  14. > 'port' = '3306',
  15. > 'username' = 'root',
  16. > 'password' = '123456',
  17. > 'database-name' = 'test',
  18. > 'table-name' = 'cdc_mysql_source'
  19. > );
  20. [INFO] Execute statement succeed.
  21. Flink SQL> CREATE TABLE cdc_doris_sink (
  22. > id INT,
  23. > name STRING,
  24. > dr TINYINT
  25. > ) WITH (
  26. > 'connector' = 'doris-inlong',
  27. > 'fenodes' = 'localhost:8030',
  28. > 'username' = 'root',
  29. > 'password' = '000000',
  30. > 'sink.enable-delete' = 'true',
  31. > 'sink.multiple.enable' = 'true',
  32. > 'sink.multiple.format' = 'canal-json',
  33. > 'sink.multiple.database-pattern' = '${database}',
  34. > 'sink.multiple.table-pattern' = 'doris_${table}'
  35. > );
  36. [INFO] Execute statement succeed.
  37. -- 支持删除事件同步(sink.enable-delete='true'), 需要 Doris 表开启批量删除功能
  38. Flink SQL> insert into cdc_doris_sink select * from cdc_mysql_source /*+ OPTIONS('server-id'='5402') */;
  39. [INFO] Submitting SQL update statement to the cluster...
  40. [INFO] SQL update statement has been successfully submitted to the cluster:
  41. Job ID: 30feaa0ede92h6b6e25ea0cfda26df5e

InLong Dashboard 用法

TODO: 将在未来支持此功能。

InLong Manager Client 用法

TODO: 将在未来支持此功能。

Doris Load 节点参数

参数是否必选默认值数据类型描述
connector必选(none)string指定要使用的连接器 doris
fenodes必选(none)stringDoris FE http 地址
table.identifier必选(none)stringDoris 表名,如:db1.tbl1
username必选(none)string访问 Doris 的用户名
password必选(none)string访问 Doris 的密码
doris.request.retries可选3int向 Doris 发送请求的重试次数
doris.request.connect.timeout.ms可选30000int向 Doris 发送请求的连接超时时间
doris.request.read.timeout.ms可选30000int向 Doris 发送请求的读取超时时间
doris.request.query.timeout.s可选3600int查询 Doris 的超时时间,默认值为1小时,-1表示无超时限制
doris.request.tablet.size可选Integer.MAX_VALUEint一个 Partition 对应的 Doris Tablet 个数。
此数值设置越小,则会生成越多的 Partition。从而提升 Flink 侧的并行度,但同时会对 Doris 造成更大的压力。
doris.batch.size可选1024int一次从 BE 读取数据的最大行数。增大此数值可减少 Flink 与 Doris 之间建立连接的次数。
从而减轻网络延迟所带来的的额外时间开销。
doris.exec.mem.limit可选2147483648long单个查询的内存限制。默认为 2GB,单位为字节
doris.deserialize.arrow.async可选falseboolean是否支持异步转换 Arrow 格式到 flink-doris-connector 迭代所需的 RowBatch
doris.deserialize.queue.size可选64int异步转换 Arrow 格式的内部处理队列,当 doris.deserialize.arrow.async 为 true 时生效
doris.read.field可选(none)string读取 Doris 表的列名列表,多列之间使用逗号分隔
doris.filter.query可选(none)string过滤读取数据的表达式,此表达式透传给 Doris。Doris 使用此表达式完成源端数据过滤。
sink.batch.size可选10000int单次写 BE 的最大行数
sink.max-retries可选1int写 BE 失败之后的重试次数
sink.batch.interval可选10sstringFlush 间隔时间,超过该时间后异步线程将缓存中数据写入 BE。 默认值为10秒,支持时间单位 ms、s、min、h和d。设置为0表示关闭定期写入。
sink.properties.可选(none)stringStream load 的导入参数

例如:
‘sink.properties.column_separator’ = ‘, ‘
定义列分隔符

‘sink.properties.escape_delimiters’ = ‘true’
特殊字符作为分隔符,’\x01’ 会被转换为二进制的 0x01

‘sink.properties.format’ = ‘json’
‘sink.properties.strip_outer_array’ = ‘true’
JSON 格式导入

‘sink.properties.format’ = ‘csv’
CSV 格式导入
sink.enable-delete可选trueboolean是否启用删除。此选项需要 Doris 表开启批量删除功能(0.15+版本默认开启),只支持 Uniq 模型。
sink.enable-delete可选trueboolean是否启用删除。此选项需要 Doris 表开启批量删除功能(0.15+版本默认开启),只支持 Uniq 模型。
sink.multiple.enable可选falseboolean是否支持 Doris 多表写入。 sink.multiple.enabletrue 时,需要 sink.multiple.formatsink.multiple.database-patternsink.multiple.table-pattern 分别设置正确的值。
sink.multiple.format可选(none)string多表写入时,表示源端二进制数据的真实格式,支持 canal-jsondebezium-json 两种格式,请参阅 kafka — 动态 Topic 提取 获取更多细节。
sink.multiple.database-pattern可选(none)string多表写入时,从源端二进制数据中按照 sink.multiple.database-pattern 指定名称提取写入的数据库名称。 sink.multiple.enable 为true时有效。
sink.multiple.table-pattern可选(none)string多表写入时,从源端二进制数据中按照 sink.multiple.table-pattern 指定名称提取写入的表名。 sink.multiple.enable 为true时有效。
sink.multiple.ignore-single-table-errors可选trueboolean多表写入时,是否忽略某个表写入失败。为 true 时,如果某个表写入异常,则不写入该表数据,其他表的数据正常写入。为 false 时,如果某个表写入异常,则所有表均停止写入。
inlong.metric.labels可选(none)Stringinlong metric 的标签值,该值的构成为groupId={groupId}&streamId={streamId}&nodeId={nodeId}
sink.multiple.schema-update.policy可选(none)string往 Doris 表同步数据时,如果 Doris 表不存在或字段长度超过限制,Doris 服务器会抛出异常。

当该属性设置为 THROW_WITH_STOP ,异常会向上抛给 Flink 框架。Flink 框架会自动重启任务,尝试恢复。

当该属性设置为 STOP_PARTIAL 时,Doris connector 会忽略该表的写入,新数据不再往该表写入,其它表则正常同步。

当该属性设置为 LOG_WITH_IGNORE 时,异常会打印到日志中,不会向上抛出。后续新数据到来时,继续尝试往该表写入。
dirty.ignore可选(none)boolean往 Doris 表同步数据时,如果遇到错误和异常,通过该变量可以控制是否忽略脏数据。如果设置为 false ,则忽略脏数据,不归档。如果为 true ,则根据其它的 dirty.side-output. 的配置决定如何归档数据。
dirty.side-output.connector可选(none)string支持 s3log 两种配置。当配置为 log 时,仅打印日志,不归档数据。当配置为 s3 时,可以将数据归档到亚马逊S3或腾讯云COS存储。
dirty.side-output.s3.bucket可选(none)stringS3 或 COS 的桶名称
dirty.side-output.s3.endpoint可选(none)stringS3 或 COS 的 endpoint 地址
dirty.side-output.s3.key可选(none)stringS3 或 COS 的 key
dirty.side-output.s3.region可选(none)stringS3 或 COS 的区域
dirty.side-output.line-delimiter可选(none)string脏数据的行分隔符
dirty.side-output.field-delimiter可选(none)string脏数据的字段分隔符
dirty.side-output.s3.secret-key-id可选(none)stringS3 或 COS 的 secret key
dirty.side-output.s3.access-key-id可选(none)stringS3 或 COS 的 access key
dirty.side-output.format可选(none)string脏数据归档的格式,支持 jsoncsv
dirty.side-output.log-tag可选(none)string脏数据的 tag 。通过该变量区分每条脏数据归属于 Doris 的哪个库表。
dirty.identifier可选(none)string归档后的文件名
dirty.side-output.labels可选(none)string归档后的每条数据包括标签和业务数据两部分。标签在前面,业务数据在后面。

数据类型映射

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

请参阅 flink-doris-connector 页面以获取更多细节。