日誌即時入湖

前提條件

如果您使用RAM使用者或RAM角色等身份訪問，需要確認已具有Flink控制台相關許可權，詳情請參見許可權管理。
已建立Flink工作空間，詳情請參見開通Realtime ComputeFlink版。
上下遊儲存
- 已建立訊息佇列Kafka執行個體，詳情請參見步驟二：購買和部署執行個體。
- 已開通資料湖構建DLF，詳情請參見授權並開通DLF。
說明
訊息佇列Kafka需要與Realtime ComputeFlink版工作空間在相同地區相同VPC下，否則需要打通網路，詳情請參見如何訪問跨VPC的其他服務？或如何訪問公網？。

步驟一：配置IP白名單

為了讓Flink能訪問Kafka執行個體，您需要將Flink工作空間的網段添加到Kafka的白名單中。

擷取Flink工作空間的VPC網段。
1. 登入Realtime Compute控制台。
2. 在目標工作空間右側操作列，選擇更多 > 工作空間詳情。
3. 在工作空間詳情對話方塊，查看虛擬交換器的網段資訊。
  
  對話方塊中展示了虛擬交換器列表，包含名稱/ID、所屬可用性區域、網段、可用IP數和可啟動TM/JM執行個體數等列。請記錄網段列中顯示的CIDR位址區段，用於後續配置Kafka白名單。
在訊息佇列Kafka的IP白名單中，添加Flink工作空間的網段資訊。

您需要為網路類型為VPC的存取點配置白名單，操作步驟請參見配置白名單。開啟編輯預設存取點白名單對話方塊後，單擊添加白名單IP按鈕添加網段。

步驟二：準備Kafka測試資料

使用Realtime ComputeFlink版的類比資料產生Faker作為資料產生器，將資料寫入到Kafka中。請按以下步驟使用Realtime Compute開發控制台將資料寫入至訊息佇列Kafka。

在Kafka控制台建立一個名稱為users的Topic。

操作詳情請參見步驟一：建立Topic。

建立將資料寫入到Kafka的作業。

登入Realtime Compute管理主控台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資料開發 > ETL。

單擊後，單擊建立流作業，填寫檔案名稱並選擇引擎版本。

Flink也為您提供了豐富的代碼模板和資料同步，每種代碼模板都為您提供了具體的使用情境、程式碼範例和使用指導。您可以直接單擊對應的模板快速地瞭解Flink產品功能和相關文法，實現您的商務邏輯，詳情請參見代碼模板和資料同步模板。

作業參數

說明

樣本

檔案名稱

作業的名稱。

說明

作業名稱在當前專案中必須保持唯一。

flink-test

引擎版本

當前作業使用的Flink引擎版本。

建議使用帶有推薦、穩定標籤的版本，這些版本具有更高的可靠性和效能表現，引擎版本詳情請參見功能發布記錄和引擎版本介紹。

vvr-11.5-jdk11-flink-1.20

單擊建立。

編寫SQL作業。

將以下作業代碼拷貝到作業文本編輯區，然後根據實際配置，修改參數配置資訊。

CREATE TEMPORARY TABLE source (
  id INT,
  first_name STRING,
  last_name STRING,
  `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
  event_time TIMESTAMP
) WITH (
  'connector' = 'faker',
  'number-of-rows' = '100',
  'rows-per-second' = '10',
  'fields.id.expression' = '#{number.numberBetween ''0'',''1000''}',
  'fields.first_name.expression' = '#{name.firstName}',
  'fields.last_name.expression' = '#{name.lastName}',
  'fields.address.country.expression' = '#{Address.country}',
  'fields.address.state.expression' = '#{Address.state}',
  'fields.address.city.expression' = '#{Address.city}',
  'fields.event_time.expression' = '#{date.past ''15'',''SECONDS''}'
);

CREATE TEMPORARY TABLE sink (
  id INT,
  first_name STRING,
  last_name STRING,
  `address` ROW<`country` STRING, `state` STRING, `city` STRING>,
  `timestamp` TIMESTAMP METADATA
) WITH (
  'connector' = 'kafka',
  'properties.bootstrap.servers' = 'alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092',
  'topic' = 'users',
  'format' = 'json',
  'properties.enable.idempotence'='false'
);

INSERT INTO sink SELECT * FROM source;

需要修改的參數配置資訊如下：

參數

樣本值

說明

properties.bootstrap.servers

alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092

Kafka Broker地址。

格式為host:port,host:port,host:port，以英文逗號（,）分割。您可以在執行個體詳情頁面的存取點資訊地區擷取網路類型為VPC的網域名稱存取點作為該參數的值。

topic

users

Kafka Topic名稱。

啟動作業。
1. 在資料開發 > ETL頁面，單擊部署。
2. 在部署新版本對話方塊中，單擊確定。
3. 配置作業資源，資源設定填寫詳情請參見配置作業資源。
4. 在營運中心 > 作業營運頁面，單擊目標作業名稱操作列中的啟動。關於作業啟動的配置說明，請參見作業啟動。
5. 您可以在作業營運頁面觀察作業的運行資訊和狀態。
  由於faker資料來源是一個有限流，因此在作業處於運行狀態後，大約1分鐘左右後，作業就會處於完成狀態。當作業結束運行代表作業已經將相關的資料寫入到Kafka的users中。其中，寫入到訊息佇列Kafka的JSON資料格式大致如下。
```
{
  "id": 765,
  "first_name": "Barry",
  "last_name": "Pollich",
  "address": {
    "country": "United Arab Emirates",
    "state": "Nevada",
    "city": "Powlowskifurt"
  }
}
```

步驟三：建立並啟動資料攝入作業

登入Realtime Compute開發控制台，建立資料攝入作業。

登入Realtime Compute管理主控台。
單擊目標工作空間操作列下的控制台。
在左側導覽列，單擊資料開發 > 資料攝入。

單擊後，單擊建立資料攝入草稿，填寫檔案名稱並選擇引擎版本。

作業參數

說明

樣本

檔案名稱

作業的名稱。

說明

作業名稱在當前專案中必須保持唯一。

flink-test

引擎版本

當前作業使用的Flink引擎版本。

建議使用帶有推薦、穩定標籤的版本，這些版本具有更高的可靠性和效能表現，引擎版本詳情請參見功能發布記錄和引擎版本介紹。

vvr-11.5-jdk11-flink-1.20

單擊建立。

編寫資料攝入作業。將以下作業代碼拷貝到作業文本編輯區，然後根據實際配置，修改參數配置資訊。

假設Kafka的topic users中存有JSON格式的表資料，下面的作業可以將表的資料同步到DLF的test_database庫的表users中。

source:
  type: kafka
  name: Kafka Source
  properties.bootstrap.servers: alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092
  topic: users
  scan.startup.mode: earliest-offset
  value.format: json
  # （可選）遞迴式地展開JSON中的嵌套列
  json.infer-schema.flatten-nested-columns.enable: true
  # （可選）跳過前 100 次出現的解析異常；若超過 100 次則作業失敗。
  ingestion.ignore-errors: true
  ingestion.error-tolerance.max-count: 100

sink:
  type: paimon
  name: Paimon Sink
  catalog.properties.metastore: rest
  catalog.properties.uri: dlf_uri
  catalog.properties.warehouse: your_warehouse
  catalog.properties.token.provider: dlf
  #（可選）開啟刪除向量，提升讀取效能
  table.properties.deletion-vectors.enabled: true

transform:
  - source-table: \.*.\.*
    projection: \*
    primary-keys: id
    
route:
  - source-table: users
    sink-table: test_database.users

pipeline:
  # （可選）將會導致處理異常的髒資料記錄到日誌中
  dirty-data.collector:
    name: Logger Dirty Data Collector
    type: logger

需要修改的參數配置資訊如下：

參數	樣本值	說明
properties.bootstrap.servers	alikafka-host1.aliyuncs.com:9092,alikafka-host2.aliyuncs.com:9092,alikafka-host3.aliyuncs.com:9092	Kafka Broker地址。格式為host:port,host:port,host:port，以英文逗號（,）分割。您可以在執行個體詳情頁面的存取點資訊地區擷取網路類型為VPC的網域名稱存取點作為該參數的值。
topic	users	Kafka Topic名稱。
catalog.properties.uri	http://cn-hangzhou-vpc.dlf.aliyuncs.com	訪問DLF Rest Catalog Server的URI，格式為`http://[region-id]-vpc.dlf.aliyuncs.com`。詳見服務存取點中的Region ID。
catalog.properties.warehouse	dlf_test	DLF Catalog名稱。
source-table	users	定義來自於上遊哪個表，預設是topic名稱。
sink-table	test_database.users	定義寫入下遊哪張表，使用逗號串連Schema和表名。

單擊儲存。
在資料開發 > 資料攝入頁面，單擊部署。
在營運中心 > 作業營運頁面，單擊目標作業名稱操作列中的啟動。關於作業啟動的配置說明，請參見作業啟動。

作業啟動後，您可以在作業營運介面觀察作業的運行資訊和狀態。

步驟四：觀察全量同步結果

登入。
選擇作業目標表所在的Catalogs。
點擊名稱為test_database的資料庫，查看users資料庫中同步的users表結構和資料。
同步後的表結構和資料如下所示。
- 表結構
  
  點擊users表名稱，查看錶結構。
  
  表結構共包含8個欄位：id（BIGINT，主鍵，非空）、first_name（STRING）、last_name（STRING）、address.country（STRING）、address.state（STRING）、address.city（STRING）、house-points.house（STRING）和house-points.points（BIGINT）。共202行資料，Snapshot ID為1。
- 表資料
  在Realtime ComputeFlink版上添加Catalog，參見建立DLF Catalog。
  
  在作業營運 > Session管理頁面，單擊建立Session叢集。
  
  在資料開發 > 資料查詢頁面，單擊建立查詢指令碼。
  
  在查詢指令碼中填寫如下內容：
  SELECT * FROM `catalog_name`.test_database.users;
  
  點擊運行，將作業指定提交到剛才建立的Session叢集中：
  
  在Realtime Compute控制台的資料查詢頁面中，左側為查詢指令碼列表，右側SQL編輯區上方有運行按鈕及格式化等工具選項。
  表資料結果如下所示。
  
  查詢結果展示了全量同步的資料記錄，包含id、first_name、last_name、address.country、address.state、address.city、house-points.house和house-points.points共8列。其中首次同步處理的記錄中house-points.house和house-points.points列的值均為空白。

步驟五：觀察自動同步表結構變更

在Kafka控制台手動發送一條包含新增列的訊息。

登入雲訊息佇列 Kafka 版控制台。
在執行個體列表頁面，單擊目標執行個體名稱。
在Topic管理頁面，單擊目標Topic名稱users。
單擊體驗發送訊息。

填寫訊息內容。

在快速體驗訊息收發對話方塊中，按照下表配置各參數。

配置項	樣本
發送方式	選中控制台。
訊息Key	填寫為flinktest。
訊息內容	將以下JSON內容複寫粘貼到訊息內容中。 `{ "id": 100001, "first_name": "Dennise", "last_name": "Schuppe", "address": { "country": "Isle of Man", "state": "Montana", "city": "East Coleburgh" }, "house-points": { "house": "Pukwudgie", "points": 76 } }` 說明該樣本中house-points是一個新增的嵌套列。
發送到指定分區	選中是。
分區ID	填寫為0。

單擊確定。

在DLF和FlinkRealtime Compute控制台，查看users表結構和資料的變化。
1. 登入。
2. 點擊名稱為test_database的資料庫，查看users資料庫中同步的users表結構和資料。
  
  可以看到users表的欄位列表中，除原有的id、first_name、last_name、address.country、address.state和address.city欄位外，新增了house-points.house（STRING）和house-points.points（BIGINT）兩個欄位，表明表結構變更已自動同步。
3. 在Realtime Compute平台資料查詢頁面執行如下查詢。
4. ```
SELECT * FROM `catalog_name`.test_database.users;
```
5. 查看錶資料結果。
  
  表資料結果如下所示。
  
  可以觀察到id為100001的資料已經成功地寫入到了DLF目標表中。同時，DLF中的test_database.users表多了house-points.house和house-points.points 兩列。
  
  說明
  雖然插入到Kafka中的資料只有一個嵌套列house-points，但是由於在users表的WITH參數內聲明要求json.infer-schema.flatten-nested-columns.enable，那麼Flink 就會自動展平新增的嵌套列，並用訪問該列的路徑作為展開後的列的名字。