数据仓库 ClickHouse

CREATE TABLE clickhouse_sink_table (
   `id` INT,
   `name` STRING
) WITH (
   -- 指定数据库连接参数
   'connector' = 'clickhouse',                       -- 指定使用clickhouse连接器
   'url' = 'clickhouse://172.28.28.160:8123',        -- 指定集群地址，可以通过ClickHouse集群界面查看
   -- 如果ClickHouse集群未配置账号密码可以不指定
   --'username' = 'root',                            -- ClickHouse集群用户名
   --'password' = 'root',                            -- ClickHouse集群的密码
   'database-name' = 'db',                           -- 数据写入目的数据库
   'table-name' = 'table',                           -- 数据写入目的数据表
   'sink.batch-size' = '1000'                        -- 触发批量写的条数
);

CREATE TABLE clickhouse_upsert_sink_table (
   `id` INT,
   `name` STRING,
  PRIMARY KEY (`id`) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
   -- 指定数据库连接参数
   'connector' = 'clickhouse',                     -- 指定使用clickhouse连接器
   'url' = 'clickhouse://172.28.28.160:8123',      -- 指定集群地址，可以通过ClickHouse集群界面查看
   -- 如果ClickHouse集群未配置账号密码可以不指定
   --'username' = 'root',                          -- ClickHouse集群用户名
   --'password' = 'root',                          -- ClickHouse集群的密码
   'database-name' = 'db',                         -- 数据写入目的数据库
   'table-name' = 'table',                         -- 数据写入目的数据表
   'table.collapsing.field' = 'Sign',              -- CollapsingMergeTree 类型列字段的名称
   'sink.batch-size' = '1000'                      -- 触发批量写的条数
);

CREATE TABLE `clickhosue_batch_source` (
    `when`     TIMESTAMP,
    `userid`   BIGINT,
    `bytes`    FLOAT
) WITH (
    'connector' = 'clickhouse',
    'url' = 'clickhouse://172.28.1.21:8123',
    'database-name' = 'dts',
    'table-name' = 'download_dist'
--    'scan.by-part.enabled' = 'false', -- 是否启用读 ClickHouse 表 part。若启用，必须先在所有节点上使用命令 'STOP MERGES' 和 'STOP TTL MERGES' 停止表的后台 merge 和基于 TTL 的数据删除操作，否则读取的数据会不正确。
--    'scan.part.modification-time.lower-bound' = '2021-09-24 16:00:00', -- 用于根据 modification_time 过滤 ClickHouse 表 part 的最小时间（包含），格式 yyyy-MM-dd HH:mm:ss。
--     'scan.part.modification-time.upper-bound' = '2021-09-17 19:16:26', -- 用于根据 modification_time 过滤 ClickHouse 表 part 的最大时间（不包含），格式 yyyy-MM-dd HH:mm:ss。
--    'local.read-write' = 'false', -- 是否读本地表，默认 false 。
--    'table.local-nodes' = '172.28.1.24:8123,172.28.1.226:8123,172.28.1.109:8123,172.28.1.36:8123' -- local node 列表，需要使用 http port。注意一个 shard 只能配置一个 replica 节点地址，否则会读取到重复数据。
);

CREATE TABLE `clickhouse_dimension` (
    `userid` BIGINT,
    `comment` STRING
) WITH (
    'connector' = 'clickhouse',
    'url' = 'clickhouse://172.28.1.21:8123',
    'database-name' = 'dimension',
    'table-name' = 'download_dist',
    'lookup.cache.max-rows' = '500', -- 查询缓存（Lookup Cache）中最多缓存的数据条数。
    'lookup.cache.ttl' = '10min', -- 查询缓存中每条记录最长的缓存时间。
    'lookup.max-retries' = '10' -- 数据库查询失败时，最多重试的次数。
);

CREATE TABLE datagen_source_table ( 
    id INT, 
    name STRING 
) WITH ( 
   'connector' = 'datagen',
   'rows-per-second'='1'  -- 每秒产生的数据条数
);
CREATE TABLE clickhouse_sink_table (
   `id` INT,
   `name` STRING
) WITH (
   -- 指定数据库连接参数
   'connector' = 'clickhouse',                       -- 指定使用clickhouse连接器
   'url' = 'clickhouse://172.28.28.160:8123',        -- 指定集群地址，可以通过ClickHouse集群界面查看
   -- 如果ClickHouse集群未配置账号密码可以不指定
   --'username' = 'root',                            -- ClickHouse集群用户名
   --'password' = 'root',                            -- ClickHouse集群的密码
   'database-name' = 'db',                           -- 数据写入目的数据库
   'table-name' = 'table',                           -- 数据写入目的数据表
   'sink.batch-size' = '1000'                        -- 触发批量写的条数
);
insert into clickhouse_sink_table select * from datagen_source_table;

CREATE TABLE testdb.testtable on cluster default_cluster (`id` Int32,`name` Nullable(String),`age` Nullable(Int32),`weight` Nullable(Float64),`Sign` Int8) ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/testdb/testtable', '{replica}', Sign) ORDER BY id SETTINGS  replicated_deduplication_window = 0;

CREATE TABLE testdb.testtable on cluster default_cluster (`id` Int32,`name` Nullable(String),`age` Nullable(Int32),`weight` Nullable(Float64),`Sign` Int8) ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/testdb/testtable', '{replica}', Sign) ORDER BY id ;

-- 创建数据库
CREATE DATABASE test ON cluster default_cluster;

-- 创建本地表
CREATE TABLE test.datagen ON cluster default_cluster (`id` Int32,`name` Nullable(String),`age` Nullable(Int32),`weight` Nullable(Float64),`Sign` Int8) ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/test/datagen', '{replica}', Sign) ORDER BY id SETTINGS replicated_deduplication_window = 0;

-- 基于本地表，创建分布式表
CREATE TABLE test.datagen_all ON CLUSTER default_cluster AS test.datagen ENGINE = Distributed(default_cluster, test, datagen, id);

-- 创建数据库
CREATE DATABASE test ON cluster default_cluster;

-- 创建本地表
CREATE TABLE test.datagen ON cluster default_cluster (`id` Int32,`name` Nullable(String),`age` Nullable(Int32),`weight` Nullable(Float64) ) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{layer}-{shard}/test/datagen', '{replica}') ORDER BY id SETTINGS replicated_deduplication_window = 0;

-- 基于本地表，创建分布式表
CREATE TABLE test.datagen_all ON CLUSTER default_cluster AS test.datagen ENGINE = Distributed(default_cluster, test, datagen, id);