数据库 Doris

CREATE TABLE doris_sink_table (
  id INT,
  name VARCHAR
) WITH (
  'connector' = 'doris',                    -- 固定值 'doris'
  'fenodes' = 'FE_IP:FE_HTTP_PORT',       -- Doris FE HTTP 地址
  'table.identifier' = 'test.sales_order',  -- Doris 表名 格式：db.tbl
  'username' = 'root',                      -- 访问Doris的用户名，拥有库的写权限
  'password' = 'password',                  -- 访问Doris的密码
  'sink.batch.size' = '500',                -- 单次写BE的最大行数
  'sink.batch.interval' = '1s'              -- flush 间隔时间，超过该时间后异步线程将 缓存中数据写入BE。 默认值为1秒，支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。
);

CREATE TABLE doris_sink_table (
  id INT,
  name VARCHAR
) WITH (
  'connector' = 'doris',                    -- 固定值 'doris'
  'fenodes' = 'FE_IP:FE_HTTP_PORT',         -- Doris FE HTTP 地址
  'table.identifier' = 'test.sales_order',  -- Doris 表名 格式：db.tbl
  'username' = 'root',                      -- 访问Doris的用户名，拥有库的写权限
  'password' = 'password'                   -- 访问Doris的密码
);

-- 注意: 默认打开 2PC 两阶段提交写入

CREATE CATALOG doris_catalog WITH (
  'type' = 'doris',
  'fenodes' = 'FE_IP:FE_HTTP_PORT',       -- Doris FE HTTP 地址
  'username' = 'root',                      -- 访问Doris的用户名，拥有库的写权限
  'password' = 'password',                  -- 访问Doris的密码
  'default-database' = 'default'
  )

----- Flink 1.13 -----
CREATE TABLE datagen_source_table ( 
    id INT,
    name STRING
) WITH ( 
  'connector' = 'datagen',
  'rows-per-second'='1'  -- 每秒产生的数据条数
);

CREATE TABLE doris_sink_table (
  id INT,
  name STRING
) WITH (
  'connector' = 'doris',                    -- 固定值 'doris'
  'fenodes' = 'FE_IP:FE_RESFUL_PORT',       -- Doris FE HTTP 地址
  'table.identifier' = 'test.sales_order',  -- Doris 表名 格式：db.tbl
  'username' = 'root',                      -- 访问Doris的用户名，拥有库的写权限
  'password' = 'password',                  -- 访问Doris的密码
  'sink.batch.size' = '500',                -- 单次写BE的最大行数
  'sink.batch.interval' = '1s'              -- flush 间隔时间，超过该时间后异步线程将 缓存中数据写入BE。 默认值为1秒，支持时间单位ms、s、min、h和d。设置为0表示关闭定期写入。
);

INSERT INTO doris_sink_table select * from datagen_source_table;


CREATE TABLE datagen_source_table ( 
    id INT,
    name STRING
) WITH ( 
  'connector' = 'datagen',
  'rows-per-second'='1'  -- 每秒产生的数据条数
);

CREATE TABLE doris_sink_table (
  id INT,
  name STRING
) WITH (
  'connector' = 'doris',
  'fenodes' = 'FE_IP:HTTP_PORT',
  'table.identifier' = 'db.table',
  'username' = 'root',
  'password' = 'password',
  'sink.label-prefix' = 'doris_label'
);

INSERT INTO doris_sink_table select * from datagen_source_table;

CREATE CATALOG doris_catalog WITH (
  'fenodes' = 'FE_IP:FE_RESFUL_PORT',       -- Doris FE HTTP 地址
  'username' = 'root',                      -- 访问Doris的用户名，拥有库的写权限
  'password' = 'password',                  -- 访问Doris的密码
  'default-database' = 'default'
);

CREATE TABLE datagen_source_table (
    id INT,
    name STRING
) WITH (
  'connector' = 'datagen',
  'rows-per-second'='1'  -- 每秒产生的数据条数
);

INSERT INTO `doris_catalog`.`my_database`.`my_table` SELECT * FROM.datagen_source_table;

--mysql cdc 源表
CREATE TABLE `mysql_cdc_source_table` (
  `id`     INT NOT NULL,
  `name`   VARCHAR,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'mysql-cdc',           -- 固定值 'mysql-cdc'
  'hostname' = 'YourHostName',         -- 数据库的 IP
  'port' = '3306',                     -- 数据库的访问端口
  'username' = 'YourUserName',         -- 数据库访问的用户名（需要提供 SHOW DATABASES、REPLICATION SLAVE、REPLICATION CLIENT、SELECT 和 RELOAD 权限）
  'password' = 'YourPassword',         -- 数据库访问的密码
  'database-name' = 'YourDatabase',    -- 需要同步的数据库
  'table-name' = 'YourTable'           -- 需要同步的数据表名     
);

--写入doris表
CREATE TABLE `print_table` (
  `id`       INT,
  `name`     STRING,
  PRIMARY KEY (`id`) NOT ENFORCED
) WITH (
  'connector' = 'doris',                    -- 固定值 'doris'
  'fenodes' = 'FE_IP:FE_RESFUL_PORT',       -- Doris FE HTTP 地址
  'table.identifier' = 'dbName.tableName',  -- Doris 表名 格式：db.tbl
  'username' = 'YourUserName',              -- 访问Doris的用户名，拥有库的写权限
  'password' = 'YourPassword',              -- 访问Doris的密码
);

insert into print_table
select id,name from mysql_cdc_source_table;

-- Uniqe 模型建表语句
CREATE TABLE `doris_sink_table` (
    `id`   int(11),
    `name` varchar(32)
) 
UNIQUE KEY(`id`) 
DISTRIBUTED BY HASH(`id`) BUCKETS 10
PROPERTIES("replication_num" = "3");

-- Aggregate 模型建表语句
CREATE TABLE `doris_sink_table` (
    `id`   int(11),
    `name` varchar(32) REPLACE DEFAULT '0'
) 
AGGREGATE KEY('id')
DISTRIBUTED BY HASH(`id`) BUCKETS 10
PROPERTIES("replication_num" = "3");  -- 注意若 BE 节点不够，会报 `Failed to find enough host in all backends` 错误，可适当减少该值。

CREATE USER 'test' IDENTIFIED BY 'test_passwd';
GRANT ALL ON test TO test;