流式计算 Flink版
流式计算 Flink版
- 文档首页
流式计算 Flink版开发参考CDC Connector 参考Doris
文档反馈
问问助手Doris Pipeline 连接器可以用作 Doris 的 Data Sink，将数据写入 Doris。 本文档介绍如何设置 Doris Pipeline 连接器。
使用限制
- Doris Pipeline 连接器暂时仅支持在 Flink 1.16-volcano 及以上引擎版本中使用。
连接器的功能
- 自动建表
- 表结构变更同步
- 数据实时同步
如何创建 Pipeline
source: type: mysql name: MySQL Source hostname: 127.0.0.1 port: 3306 username: admin password: pass tables: adb.\.*, bdb.user_table_[0-9]+, [app|web].order_\.* server-id: 5401-5404 sink: type: doris name: Doris Sink fenodes: 127.0.0.1:8030 username: root password: "" table.create.properties.replication_num: 1 pipeline: parallelism: 1
连接器配置项
Option | Required | Default | Type | Description |
|---|---|---|---|---|
type | required | (none) | String | 指定要使用的Sink, 这里是 'doris'. |
name | optional | (none) | String | PipeLine的名称 |
fenodes | required | (none) | String | Doris集群FE的Http地址, 比如 127.0.0.1:8030 |
benodes | optional | (none) | String | Doris集群BE的Http地址, 比如 127.0.0.1:8040 |
jdbc-url | optional | (none) | String | Doris集群的JDBC地址，比如：jdbc:mysql://127.0.0.1:9030/db |
username | required | (none) | String | Doris集群的用户名 |
password | optional | (none) | String | Doris集群的密码 |
auto-redirect | optional | FALSE | String | 是否通过FE重定向写入，直连BE写入 |
sink.enable.batch-mode | optional | TRUE | Boolean | 是否使用攒批方式写入Doris |
sink.flush.queue-size | optional | 2 | Integer | 攒批写入的队列大小 |
sink.buffer-flush.max-rows | optional | 50000 | Integer | 单个批次最大Flush的记录数 |
sink.buffer-flush.max-bytes | optional | 10485760(10MB) | Integer | 单个批次最大Flush的字节数 |
sink.buffer-flush.interval | optional | 10s | String | Flush的间隔时长，超过这个时间，将异步Flush数据 |
sink.properties. | optional | (none) | String | StreamLoad的参数。 For example: sink.properties.strict_mode: true. 查看更多关于 StreamLoad的Properties 属性 |
table.create.properties.* | optional | (none) | String | 创建表的Properties配置。 For example: table.create.properties.replication_num: 1. 查看更多关于 Doris Table 的 Properties 属性 |
数据类型映射
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) | 在Doris中，字符串是以UTF-8编码存储的，所以英文字符占1个字节，中文字符占3个字节。这里的长度统一乘3，CHAR最大的长度是255，超过后会自动转为VARCHAR类型 |
VARCHAR(n) | VARCHAR(n*3) | 同上，这里的长度统一乘3，VARCHAR最大的长度是65533，超过后会自动转为STRING类型 |
BINARY(n) | STRING | |
VARBINARY(N) | STRING | |
STRING | STRING |