Tablestore：使用教程（寬表模型）

DDL定義

資料來源表的DDL定義樣本如下：

CREATE TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR
) WITH (
    'connector'='ots',
    'endPoint' ='https://flink-source.cn-hangzhou.ots.aliyuncs.com',
    'instanceName' = 'flink-source',
    'tableName' ='flink_source_table',
    'tunnelName' = 'flinksourcestream',
    'accessId' ='xxxxxxxxxxx',
    'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
    'ignoreDelete' = 'false' //是否忽略delete操作的資料.
);

除了待消費的使用者資料外，Tunnel Service返回資料中的OtsRecordType、OtsRecordTimestamp欄位均支援通過屬性欄位的方式讀取。欄位說明請參見下表。

當需要讀取OtsRecordType和OtsRecordTimestamp欄位時，Flink提供了METADATA關鍵字用於擷取源表中的屬性欄位，具體DDL樣本如下：

CREATE TABLE tablestore_stream(
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    record_type STRING METADATA FROM 'type',
    record_timestamp BIGINT METADATA FROM 'timestamp'
) WITH (
    ...
);

WITH參數

DDL定義

結果表的DDL定義樣本如下：

CREATE TABLE ots_sink (
    `order` VARCHAR,
    orderid VARCHAR,
    customerid VARCHAR,
    customername VARCHAR,
    PRIMARY KEY (`order`,orderid) NOT ENFORCED
) WITH (
     ...
);

WITH參數

結果表欄位類型映射

前提條件

• 已建立AccessKey。具體操作，請參見建立AccessKey。

• 已為Tablestore資料表（源表）建立資料通道。具體操作，請參見建立資料通道。

步驟一：建立作業

1. 登入Realtime Compute控制台。

2. 在Flink全託管頁簽，單擊目標工作空間操作列下的控制台。

3. 在左側導覽列，單擊SQL開發。

4. 單擊建立。

5. 在新增作業草稿對話方塊中，單擊空白的流作業草稿。

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

6. 單擊下一步。

7. 填寫作業配置資訊。

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

8. 單擊建立。

步驟二：編寫作業代碼

1. 建立一個源表和結果表的暫存資料表。

   說明

   在生產作業中，建議您盡量減少暫存資料表的使用，直接使用中繼資料管理中已經註冊的表。

   建立一個tablestore_stream和ots_sink暫存資料表程式碼範例如下：

   CREATE TEMPORARY TABLE tablestore_stream(
       `order` VARCHAR,
       orderid VARCHAR,
       customerid VARCHAR,
       customername VARCHAR
   ) WITH (
       'connector'='ots',
       'endPoint' ='https://flink-source.cn-hangzhou.ots.aliyuncs.com',
       'instanceName' = 'flink-source',
       'tableName' ='flink_source_table',
       'tunnelName' = 'flinksourcestream',
       'accessId' ='xxxxxxxxxxx',
       'accessKey' ='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
       'ignoreDelete' = 'false' //是否忽略delete操作的資料。
   );
   
   CREATE TEMPORARY TABLE ots_sink (
       `order` VARCHAR,
       orderid VARCHAR,
       customerid VARCHAR,
       customername VARCHAR,
       PRIMARY KEY (`order`,orderid) NOT ENFORCED
   ) WITH (
       'connector'='ots',
       'endPoint'='https://flink-sink.cn-hangzhou.ots.aliyuncs.com',
       'instanceName'='flink-sink',
       'tableName'='flink_sink_table',
       'accessId'='xxxxxxxxxxx',
       'accessKey'='xxxxxxxxxxxxxxxxxxxxxxxxxxxx',
       'valueColumns'='customerid,customername'
   );

2. 編寫作業邏輯。

   將源表資料插入到結果表的程式碼範例如下：

   INSERT INTO ots_sink
   SELECT `order`, orderid, customerid, customername FROM tablestore_stream;

步驟三：進行更多配置

在作業開發頁面右側，單擊更多配置後，您可以填寫以下參數資訊：

• 引擎版本：修改您建立作業時選擇的Flink引擎版本。

  說明 從VVR 3.0.3版本（對應Flink 1.12版本）開始，VVP支援同時運行多個不同引擎版本的SQL作業。如果您的作業已使用了Flink 1.12及更早版本的引擎，您需要按照以下情況進行處理：

  • Flink 1.12版本：停止後啟動作業，系統將自動將引擎升級為vvr-3.0.3-flink-1.12版本。
  • Flink 1.11或Flink 1.10版本：手動將作業引擎版本升級到vvr-3.0.3-flink-1.12或vvr-4.0.7-flink-1.13版本後重啟作業，否則會在啟動作業時逾時報錯。

• 附加依賴檔案：作業中需要使用到的附加依賴，例如臨時函數等。

步驟五：（可選）進行作業調試

在作業開發頁面頂部，單擊調試。

您可以使用作業調試功能類比作業運行、檢查輸出結果，驗證SELECT或INSERT商務邏輯的正確性，提升開發效率，降低資料品質風險。具體操作，請參見作業調試。

步驟六：作業部署

在作業開發頁面頂部，單擊部署，在部署新版本對話方塊，可根據需要填寫或選中相關內容，單擊確認。

步驟七：啟動並查看Flink計算結果

1. 在左側導覽列，單擊作業營運。

2. 單擊目標作業名稱操作列中的啟動。

   作業啟動參數配置詳情請參見作業啟動。單擊啟動後，您可以看到作業狀態變為運行中，則代表作業運行正常。

3. 在作業營運詳情頁面，查看Flink計算結果。

   1. 在作業營運頁面，單擊目標作業名稱。

   2. 單擊作業探查。

   3. 在作業記錄頁簽，單擊運行Task Managers頁簽下的Path, ID。

   4. 單擊日誌，在頁面搜尋Sink相關的日誌資訊。