從Flink SQL匯入 - ApsaraDB for ClickHouse

本文介紹如何使用Flink SQL方式將Flink中的資料寫入雲資料庫ClickHouse。

使用限制

只有Flink計算引擎VVR 3.0.2及以上版本支援使用Flink SQL寫入雲資料庫ClickHouse。

前提條件

已在雲資料庫ClickHouse中建立表。更多資訊，請參見建立表。
已在雲資料庫ClickHouse中設定白名單。更多資訊，請參見設定白名單。
已開通Flink全託管。更多資訊，請參見開通Realtime ComputeFlink版。

操作步驟

登入Flink全託管控制台，新增作業。

登入Realtime Compute控制台。
在Flink全託管頁簽，單擊目標工作空間操作列下的控制台。
在SQL開發頁面，單擊建立。
單擊空白的流作業草稿。
Flink全託管為您提供了豐富的SQL代碼模板和資料同步模板，每種模板都為您提供了使用情境、程式碼範例和使用指導。您也可以直接單擊對應的模板進行使用，快速實現您的商務邏輯，詳情請參見代碼模板和資料同步模板。
單擊下一步。

在新增作業草稿對話方塊，填寫作業配置資訊。

作業參數	說明
檔案名稱	作業的名稱。說明作業名稱在當前專案中必須保持唯一。
儲存位置	指定該作業的儲存位置。您還可以在現有檔案夾右側，單擊表徵圖，建立子檔案夾。
引擎版本	當前作業使用的Flink的引擎版本。引擎版本號碼含義、版本對應關係和生命週期重要時間點詳情請參見引擎版本介紹。

單擊建立。

編寫並執行代碼。

建立源表、結果表，並將源表資料插入到結果表中。

--建立源表sls_test_single_local。
CREATE TEMPORARY TABLE sls_test_single_local (
  id INT,
  name VARCHAR,
  age BIGINT,
  rate FLOAT
) WITH (
  'connector' = 'datagen',
  'rows-per-second' = '50'
);
--建立結果表clickhouse_output。
CREATE TEMPORARY TABLE clickhouse_output (
  id INT,
  name VARCHAR,
  age BIGINT,
  rate FLOAT
) WITH (
  'connector' = 'clickhouse',
  'url' = 'jdbc:clickhouse://demo.aliyuncs.com:8123',
  'userName' = 'test',
  'password' = '280226Ck',
  'tableName' = 'sls_test_single_local'
);
--將源表資料插入到結果表。
INSERT INTO clickhouse_output
SELECT 
  id,
  name,
  age,
  rate
FROM sls_test_single_local;

說明

建立結果表的文法說明，請參見ClickHouse結果表。

單擊儲存。
單擊驗證。
單擊上線。
在彈出的上線確認視窗中單擊確認，將作業發布至生產環境。
上線成功後，在彈出的上線成功，請前往營運查看詳情中單擊營運。

在作業營運頁面，啟動作業。
1. 單擊右上方啟動。
2. 在彈出的作業啟動配置頁面，單擊確認啟動。
在雲資料庫ClickHouse中查詢表。
1. 登入雲資料庫ClickHouse控制台。
2. 在頁面左上方，選擇目的地組群所在的地區。
3. 在叢集列表頁面，選擇目的地組群對應類型的執行個體列表，單擊目的地組群ID。
4. 在叢集資訊頁面，單擊右上方導覽列的登入資料庫。
5. 在登入執行個體頁面，輸入資料庫帳號和密碼，單擊登入。
6. 輸入查詢語句並單擊執行（F8）。本文使用下面的樣本語句。
```
select * from db01.sls_test_single_local;
```
  查詢結果如下。