Flink 实践教程：入门5-写入 ClickHouse

 # 下载 ClickHouse-Client 命令
wget https://repo.yandex.ru/clickhouse/rpm/stable/x86\_64/clickhouse-client-20.7.2.30-2.noarch.rpm
wget https://repo.yandex.ru/clickhouse/rpm/stable/x86\_64/clickhouse-common-static-20.7.2.30-2.x86\_64.rpm

rpm -ivh \*.rpm

clickhouse-client -hxxx.xxx.xxx.xxx --port 9000

CREATE TABLE default.datagen_to_ck on cluster default_cluster (
win_start     TIMESTAMP,
win_end       TIMESTAMP,
user_id       String,
amount_total  Int16,
Sign          Int8  )
ENGINE = ReplicatedCollapsingMergeTree('/clickhouse/tables/{layer}-{shard}/default/datagen_to_ck', '{replica}',Sign)
ORDER BY (win_start,win_end,user_id);

CREATE TABLE random_source ( 
    user_id   VARCHAR,
    amount    INT,
    pre_time  AS CURRENT_TIMESTAMP,
    WATERMARK FOR pre_time AS pre_time - INTERVAL '3' SECOND
  ) WITH ( 
  'connector' = 'datagen', 
  'rows-per-second' = '5',            -- 每秒产生的数据条数
  'fields.user_id.length' = '1',      -- 随机字符串的长度
  'fields.amount.kind' = 'random',    -- 无界的随机数
  'fields.amount.min' = '1',          -- 随机数的最小值
  'fields.amount.max' = '10'          -- 随机数的最大值
);

CREATE TABLE clickhouse (
    win_start     TIMESTAMP(3),
    win_end       TIMESTAMP(3),
    user_id       VARCHAR,
    amount_total  BIGINT,
    PRIMARY KEY (win_start,win_end,user_id) NOT ENFORCED -- 如果要同步的数据库表定义了主键, 则这里也需要定义
) WITH (
    'connector' = 'clickhouse',
    'url' = 'clickhouse://10.0.0.178:8123',
    --'username' = 'root',     -- 如果ClickHouse集群未配置账号密码可以不指定
    --'password' = 'root',
    'database-name' = 'default',
    'table-name' = 'datagen_to_ck',
    'table.collapsing.field' = 'Sign'   -- CollapsingMergeTree 类型列字段的名称
);

INSERT INTO clickhouse
SELECT
    TUMBLE_START(pre_time,INTERVAL '1' MINUTE) AS win_start,
    TUMBLE_END(pre_time,INTERVAL '1' MINUTE) AS win_end,
    user_id,
    CAST(SUM(amount) AS BIGINT) AS amount_total
FROM random_source
GROUP BY TUMBLE(pre_time,INTERVAL '1' MINUTE),user_id;