功能描述

DLI支持將Flink作業(yè)數(shù)據(jù)輸出到ClickHouse 數(shù)據(jù)庫 中，表類型僅支持結(jié)果表。

ClickHouse是面向聯(lián)機分析處理的列式數(shù)據(jù)庫，支持SQL查詢，且查詢性能好，特別是基于大寬表的聚合分析查詢性能非常優(yōu)異，比其他分析型數(shù)據(jù)庫速度快一個數(shù)量級。詳細請參考ClickHouse組件操作。

前提條件

• 該場景作業(yè)需要運行在DLI的獨享隊列上。
• 該場景需要與ClickHouse建立增強型跨源連接，并根據(jù)實際情況設(shè)置ClickHouse集群所在安全組規(guī)則中的端口。

  如何建立增強型跨源連接，請參考《數(shù)據(jù)湖探索用戶指南》中增強型跨源連接章節(jié)。

  如何設(shè)置安全組規(guī)則，請參見《 虛擬私有云 用戶指南》中“安全組”章節(jié)。

注意事項

• 創(chuàng)建Flink OpenSource SQL作業(yè)時，在作業(yè)編輯界面的“運行參數(shù)”處，“Flink版本”需要選擇“1.15”，勾選“保存作業(yè)日志”并設(shè)置保存作業(yè)日志的 OBS 桶，方便后續(xù)查看作業(yè)日志。
• 認證用的username和password等硬編碼到代碼中或者明文存儲都有很大的安全風險，建議使用 DEW 管理憑證。配置文件或者環(huán)境變量中密文存放，使用時解密，確保安全。Flink Opensource SQL使用DEW管理訪問憑據(jù)
• 創(chuàng)建 MRS 的ClickHouse集群，集群版本選擇MRS 3.1.0及以上版本。
• ClickHouse結(jié)果表不支持刪除表數(shù)據(jù)操作。
• Flink中支持字段類型范圍為：string、tinyint、smallint、int、bigint、float、double、date、timestamp、decimal以及Array。

  其中Array中的數(shù)據(jù)類型僅支持int、bigint、string、float、double。

語法格式

123456789

create table clickhouseSink (  attr_name attr_type   (',' attr_name attr_type)* )with (  'type' = 'clickhouse',  'url' = '',  'table-name' = '');

參數(shù)說明

示例

• 示例1：從Kafka中讀取數(shù)據(jù)，并將數(shù)據(jù)插入ClickHouse中（ClickHouse版本為MRS的21.3.4.25，且MRS集群未開啟Kerberos認證）：
  1. 參考增強型跨源連接，在DLI上根據(jù)ClickHouse和Kafka集群所在的虛擬 私有云 和子網(wǎng)分別創(chuàng)建跨源連接，并綁定所要使用的Flink彈性資源池。
  2. 設(shè)置ClickHouse和Kafka集群安全組的入向規(guī)則，使其對當前將要使用的Flink作業(yè)隊列網(wǎng)段放通。參考測試地址連通性根據(jù)ClickHouse和Kafka的地址測試隊列連通性。如果能連通，則表示跨源已經(jīng)綁定成功，否則表示未成功。
  3. 使用ClickHouse客戶端連接到ClickHouse服務端，并使用以下命令查詢集群標識符cluster等其他環(huán)境參數(shù)信息。

     詳細操作請參考從零開始使用ClickHouse。

     select cluster,shard_num,replica_num,host_name from system.clusters;

     其返回信息如下圖：

     ┌─cluster────┬────┬─shard_num─┐│ default_cluster │    1   │           1 ││ default_cluster │    1   │           2 │└──────── ┴────┴────── ┘

     根據(jù)獲取到的集群標識符cluster，例如當前為default_cluster ，使用以下命令在ClickHouse的default_cluster集群節(jié)點上創(chuàng)建數(shù)據(jù)庫flink。

     CREATE DATAB AS E flink ON CLUSTER default_cluster;

  4. 使用以下命令在default_cluster集群節(jié)點上和flink數(shù)據(jù)庫下創(chuàng)建表名為order的ReplicatedMergeTree表。

     CREATE TABLE flink.order ON CLUSTER default_cluster(order_id String,order_channel String,order_time String,pay_amount Float64,real_pay Float64,pay_time String,user_id String,user_name String,area_id String) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/flink/order', '{replica}')ORDER BY order_id;

  5. 參考創(chuàng)建Flink OpenSource作業(yè)，創(chuàng)建flink opensource sql作業(yè)，輸入以下作業(yè)腳本，并提交運行。該作業(yè)腳本將DMS Kafka作為數(shù)據(jù)源，ClickHouse作業(yè)結(jié)果表。

     如下腳本中的加粗參數(shù)請根據(jù)實際環(huán)境修改。

     create table orders (  order_id string,  order_channel string,  order_time string,  pay_amount double,  real_pay double,  pay_time string,  user_id string,  user_name string,  area_id string) WITH (  'connector' = 'kafka',  'topic' = 'KafkaTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  'properties.group.id' = 'GroupId',  'scan.startup.mode' = 'latest-offset',  'format' = 'json');create table clickhouseSink(  order_id string,  order_channel string,  order_time string,  pay_amount double,  real_pay double,  pay_time string,  user_id string,  user_name string,  area_id string) with (  'connector' = 'clickhouse',  'url' = 'jdbc:clickhouse://ClickhouseAddress1:ClickhousePort,ClickhouseAddress2:ClickhousePort/flink',  'username' = 'username',  'password' = 'password',  'table-name' = 'order',  'sink.buffer-flush.max-rows' = '10',  'sink.buffer-flush.interval' = '3s');insert into clickhouseSink select * from orders;

  6. 連接Kafka集群，向DMS Kafka中插入以下測試數(shù)據(jù)：

     {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}{"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}{"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}

  7. 使用ClickHouse客戶端連接到ClickHouse，執(zhí)行以下查詢命令，查詢寫入flink數(shù)據(jù)庫下order表中的數(shù)據(jù)。

     select * from flink.order;

     查詢結(jié)果參考如下：

     202103241000000001 webShop 2021-03-24 10:00:00 100 100 2021-03-24 10:02:03 0001 Alice 330106202103241606060001 appShop 2021-03-24 16:06:06 200 180 2021-03-24 16:10:06 0001 Alice 330106 202103251202020001 miniAppShop 2021-03-25 12:02:02 60 60 2021-03-25 12:03:00 0002 Bob 330110 

• 示例2：從Kafka中讀取數(shù)據(jù)，并將數(shù)據(jù)插入ClickHouse中（ClickHouse版本為MRS的21.3.4.25，且MRS集群開啟Kerberos認證）
  1. 參考增強型跨源連接，在DLI上根據(jù)ClickHouse和Kafka集群所在的虛擬私有云和子網(wǎng)分別創(chuàng)建跨源連接，并綁定所要使用的Flink彈性資源池。
  2. 設(shè)置ClickHouse和Kafka集群安全組的入向規(guī)則，使其對當前將要使用的Flink作業(yè)隊列網(wǎng)段放通。參考測試地址連通性根據(jù)ClickHouse和Kafka的地址測試隊列連通性。如果能連通，則表示跨源已經(jīng)綁定成功，否則表示未成功。
  3. 使用ClickHouse客戶端連接到ClickHouse服務端，并使用以下命令查詢集群標識符cluster等其他環(huán)境參數(shù)信息。

     參考從零開始使用ClickHouse。

     select cluster,shard_num,replica_num,host_name from system.clusters;

     其返回信息如下圖：

     ┌─cluster────┬────┬─shard_num─┐│ default_cluster │    1   │           1 ││ default_cluster │    1   │           2 │└──────── ┴────┴────── ┘

     根據(jù)獲取到的集群標識符cluster，例如當前為default_cluster ，使用以下命令在ClickHouse的default_cluster集群節(jié)點上創(chuàng)建數(shù)據(jù)庫flink。

     CREATE DATABASE flink ON CLUSTER default_cluster;

  4. 使用以下命令在default_cluster集群節(jié)點上和flink數(shù)據(jù)庫下創(chuàng)建表名為order的ReplicatedMergeTree表。

     CREATE TABLE flink.order ON CLUSTER default_cluster(order_id String,order_channel String,order_time String,pay_amount Float64,real_pay Float64,pay_time String,user_id String,user_name String,area_id String) ENGINE = ReplicatedMergeTree('/clickhouse/tables/{shard}/flink/order', '{replica}')ORDER BY order_id;

  5. 參考創(chuàng)建Flink OpenSource作業(yè)，創(chuàng)建flink opensource sql作業(yè)，輸入以下作業(yè)腳本，并提交運行。該作業(yè)腳本將Kafka作為數(shù)據(jù)源，ClickHouse作業(yè)結(jié)果表。

     如下腳本中的加粗參數(shù)請根據(jù)實際環(huán)境修改。

     create table orders (  order_id string,  order_channel string,  order_time string,  pay_amount double,  real_pay double,  pay_time string,  user_id string,  user_name string,  area_id string) WITH (  'connector' = 'kafka',  'topic' = 'KafkaTopic',  'properties.bootstrap.servers' = 'KafkaAddress1:KafkaPort,KafkaAddress2:KafkaPort',  'properties.group.id' = 'GroupId',  'scan.startup.mode' = 'latest-offset',  'format' = 'json');create table clickhouseSink(  order_id string,  order_channel string,  order_time string,  pay_amount double,  real_pay double,  pay_time string,  user_id string,  user_name string,  area_id string) with (  'connector' = 'clickhouse',  'url' = 'jdbc:clickhouse://ClickhouseAddress1:ClickhousePort,ClickhouseAddress2:ClickhousePort/flink?ssl=true&sslmode=none',  'table-name' = 'order',  'username' = 'username',  'password' = 'password', --DEW憑據(jù)中的key  'sink.buffer-flush.max-rows' = '10',  'sink.buffer-flush.interval' = '3s',  'dew.endpoint'='kms.xx.xx.com', --使用的DEW服務所在的endpoint信息  'dew.csms.secretName'='xx', --DEW服務通用憑據(jù)的憑據(jù)名稱  'dew.csms.decrypt.fields'='password', --password字段值需要利用DEW憑證管理,進行解密替換  'dew.csms.version'='v1');insert into clickhouseSink select * from orders;

  6. 連接Kafka集群，向Kafka中插入以下測試數(shù)據(jù)：

     {"order_id":"202103241000000001", "order_channel":"webShop", "order_time":"2021-03-24 10:00:00", "pay_amount":"100.00", "real_pay":"100.00", "pay_time":"2021-03-24 10:02:03", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}{"order_id":"202103241606060001", "order_channel":"appShop", "order_time":"2021-03-24 16:06:06", "pay_amount":"200.00", "real_pay":"180.00", "pay_time":"2021-03-24 16:10:06", "user_id":"0001", "user_name":"Alice", "area_id":"330106"}{"order_id":"202103251202020001", "order_channel":"miniAppShop", "order_time":"2021-03-25 12:02:02", "pay_amount":"60.00", "real_pay":"60.00", "pay_time":"2021-03-25 12:03:00", "user_id":"0002", "user_name":"Bob", "area_id":"330110"}

  7. 使用ClickHouse客戶端連接到ClickHouse，執(zhí)行以下查詢命令，查詢寫入flink數(shù)據(jù)庫下order表中的數(shù)據(jù)。

     select * from flink.order;

     查詢結(jié)果參考如下：

     202103241000000001 webShop 2021-03-24 10:00:00 100 100 2021-03-24 10:02:03 0001 Alice 330106202103241606060001 appShop 2021-03-24 16:06:06 200 180 2021-03-24 16:10:06 0001 Alice 330106 202103251202020001 miniAppShop 2021-03-25 12:02:02 60 60 2021-03-25 12:03:00 0002 Bob 330110