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。

• 已為表格存儲數據表（源表）創建數據通道。具體操作，請參見創建數據通道。

步驟一：創建作業

1. 登錄實時計算控制臺。

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相關的日志信息。