Doris 连接器

简述

• 支持 Doris 数据库的读写
• Flink Doris Connector 官方文档：https://doris.apache.org/zh-CN/docs/1.2/ecosystem/flink-doris-connector
• 插件 1.4.12+ 版本开始可用
• 需要使用 Flink 1.14 引擎，不支持 Flink 1.10/1.12

通过 SQL 读取

CREATE TABLE flink_doris_source (
    name STRING,
    age INT,
    price DECIMAL(5,2),
    sale DOUBLE
)  WITH (
      'connector' = 'doris',
      'fenodes' = 'FE_IP:8030',
      'table.identifier' = 'db.table',
      'username' = 'root',
      'password' = 'password'
);

通过 SQL 写入

注意：更新和删除数据需要基于 Unique Key 模型

-- enable checkpoint
SET 'execution.checkpointing.interval' = '60s';
CREATE TABLE flink_doris_sink (
    name STRING,
    age INT,
    price DECIMAL(5,2),
    sale DOUBLE
) WITH (
      'connector' = 'doris',
      'fenodes' = 'FE_IP:8030',
      'table.identifier' = 'db.table',
      'username' = 'root',
      'password' = 'password',
      'sink.label-prefix' = 'doris_label'
);

INSERT INTO flink_doris_sink select name,age,price,sale from flink_doris_source;

通用配置项