通過Tablestore觸發器對錶格儲存的資料進行自訂處理 - Function Compute

阿里雲Table StoreTablestore是構建在阿里雲飛天分布式系統之上的分布式NoSQL資料存放區服務。您可以通過建立Tablestore觸發器，將Tablestore作為事件來源接入Function Compute，當Tablestore中的資料更新時會自動觸發函數執行，從而完成對Tablestore變更資料的自訂處理。

使用情境

Tablestore觸發器典型的使用情境如下圖所示。

原始訊號來源資料儲存到原始Table A，當Table A中的資料發生變更時會觸發函數自訂清洗資料，將清洗後的資料存入Table B，您可以直接讀取Table B的資料統計展示，完成一個彈性可伸縮的Serverless Web應用。

前提條件

Function Compute
- 建立函數
Table StoreTablestore
- 建立Tablestore執行個體
- 建立資料表

使用限制

目前支援Tablestore觸發器的地區為：華北2（北京）、華東1（杭州）、華東2（上海）、華南1（深圳）、日本（東京）、新加坡、德國（法蘭克福）、中國香港。
Table Store資料表與Function Compute服務要處於同一地區。
若您需要使用內網訪問Tablestore觸發器對應的函數，可以使用VPC Endpoint：{instance}.{region}.vpc.tablestore.aliyuncs.com，不可以使用Tablestore內網Endpoint。
觸發的函數執行時間不能超過一分鐘。

注意事項

編寫函數的時候，請注意不要出現以下邏輯：Table StoreTable A觸發函數B，函數B又更新Table A的資料，從而造成函數無限迴圈調用。
若函數執行出現異常，函數將無限重試直到Tablestore中的日誌資料到期。
說明
- 函數執行異常有以下情況：
  - 函數代碼運行異常：函數執行個體已拉起，因此函數執行個體啟動並執行時間段內會有費用產生。
  - 函數啟動異常：函數執行個體由於啟動指令錯誤等原因未成功拉起，此時不會有費用產生。
- 如果函數執行異常，為避免函數無限重試，您可以關閉資料表的Stream功能。在關閉資料表的Stream功能前請確認沒有其他觸發器在使用該資料表，防止導致其他觸發器異常。

步驟一：為資料表開啟Stream功能

使用觸發器功能需要先在Table Store控制台開啟資料表的Stream功能，才能在Function Compute中處理寫入Table Store中的增量資料。

登入Table Store控制台。
在頁面上方，選擇地區。
在概覽頁面，單擊執行個體名稱或在執行個體操作列單擊執行個體管理。
在執行個體詳情頁簽的資料表列表頁簽，單擊資料表名稱後選擇即時消費通道頁簽或單擊後選擇即時消費通道。
在即時消費通道頁簽，單擊Stream資訊對應的開啟。
在開啟Stream功能對話方塊，設定日誌到期時間長度，單擊開啟。
日誌到期時間長度取值為非零整數，單位為小時，最長時間長度為168小時。
重要
日誌到期時間長度設定後不能修改，請謹慎設定。

步驟二：建立Tablestore觸發器

登入Function Compute控制台，在左側導覽列，單擊函數。
在頂部功能表列，選擇地區，然後在函數頁面，單擊目標函數。
在函數詳情頁面，選擇配置頁簽，在左側導覽列，單擊觸發器，然後單擊建立觸發器。

在建立觸發程序面板，填寫相關資訊，然後單擊確定。

配置項	操作	本文樣本
觸發器類型	選擇Table Store Tablestore。	Table StoreTablestore
名稱	自訂填寫觸發器名稱。	Tablestore-trigger
版本或別名	預設值為LATEST，如果您需要建立其他版本或別名的觸發器，需先在函數詳情頁的版本或別名下拉式清單選擇該版本或別名。關於版本和別名的簡介，請參見版本管理和別名管理。	LATEST
執行個體	在列表中選擇已建立的Tablestore執行個體。	d00dd8xm****
表格	在列表中選擇已建立的表格。	mytable
角色名稱	選擇AliyunTableStoreStreamNotificationRole。說明如果您第一次建立該類型的觸發器，則需要在單擊確定後，在彈出的對話方塊中選擇立即授權。	AliyunTableStoreStreamNotificationRole

建立完成後，在觸發器名稱列表中顯示已建立的觸發器。如需對建立的觸發器進行修改或刪除，具體操作，請參見觸發器管理。

步驟三：配置函數的入口參數

在函數詳情頁面的代碼頁簽，單擊測試函數右側的表徵圖，從下拉式清單中，選擇配置測試參數。

在配置測試參數面板，選擇建立新測試事件或編輯已有測試事件，填寫事件名稱和事件內容，然後單擊確定。

Table Store觸發器使用CBOR格式對增量資料進行編碼，構成Function Compute的event。具體格式如下所示。

{
    "Version": "Sync-v1",
    "Records": [
        {
            "Type": "PutRow",
            "Info": {
                "Timestamp": 1506416585740836
            },
            "PrimaryKey": [
                {
                    "ColumnName": "pk_0",
                    "Value": 1506416585881590900
                },
                {
                    "ColumnName": "pk_1",
                    "Value": "2017-09-26 17:03:05.8815909 +0800 CST"
                },
                {
                    "ColumnName": "pk_2",
                    "Value": 1506416585741000
                }
            ],
            "Columns": [
                {
                    "Type": "Put",
                    "ColumnName": "attr_0",
                    "Value": "hello_table_store",
                    "Timestamp": 1506416585741
                },
                {
                    "Type": "Put",
                    "ColumnName": "attr_1",
                    "Value": 1506416585881590900,
                    "Timestamp": 1506416585741
                }
            ]
        }
    ]
}

event參數中不同屬性欄位的解釋如下表所示。

參數	描述
Version	Payload版本號碼。樣本如Sync-v1。類型為String。
Records	資料表中的增量資料行數組。包含如下內部成員： Type：資料行類型，包含PutRow、UpdateRow和DeleteRow。類型為String。 Info：包含Timestamp內部成員。Timestamp表示該行的最後修改UTC時間。類型為Int64。
PrimaryKey	主鍵列數組。包含如下內部成員： ColumnName：主鍵列名稱。類型為String。 Value：主鍵列內容。類型為formated_value，支援Integer、String和Blob。
Columns	屬性列數組。包括如下內部成員： Type：屬性列類型，包含Put、DeleteOneVersion和DeleteAllVersions。類型為String。 ColumnName：屬性列名稱。類型為String。 Value：屬性列內容。類型為formated_value，支援Integer、Boolean、Double、String和Blob。 Timestamp：屬性列最後修改UTC時間。類型為Int64。

步驟四：編寫函數代碼並測試

完成Tablestore觸發器建立後，您可以開始編寫函數代碼並測試，以驗證代碼的正確性。在實際操作過程中當Tablestore中有資料更新時會自動觸發函數執行。

在函數詳情頁面的代碼頁簽，在代碼編輯器中編寫代碼，然後單擊部署代碼。

本文以Python函數代碼為例。如果您想使用其他運行環境，更多程式碼範例，請參見Table Store觸發Function Compute樣本。

import logging
import cbor
import json
 def get_attribute_value(record, column):
     attrs = record[u'Columns']
     for x in attrs:
         if x[u'ColumnName'] == column:
             return x['Value']
 def get_pk_value(record, column):
     attrs = record[u'PrimaryKey']
     for x in attrs:
         if x['ColumnName'] == column:
             return x['Value']
 def handler(event, context):
     logger = logging.getLogger()
     logger.info("Begin to handle event")
     #records = cbor.loads(event)
     records = json.loads(event)
     for record in records['Records']:
         logger.info("Handle record: %s", record)
         pk_0 = get_pk_value(record, "pk_0")
         attr_0 = get_attribute_value(record, "attr_0")
     return 'OK'

單擊測試函數。
執行完成後，您可以在函數代碼頁簽的上方查看執行結果。

常見問題

如果您無法在某一地區建立Tablestore觸發器，請確認支援建立Tablestore觸發器的地區，具體請查看使用限制。
如果您在建立Tablestore觸發器時無法找到已經建立好的Table Store資料表，請確認Table Store資料表與Function Compute服務是否處於同一地區。
使用Tablestore觸發器時，總是會報用戶端取消的報錯，一般是由於用戶端調用函數時設定的逾時時間小於函數執行時間。建議您將用戶端逾時時間調大，具體請參見用戶端中斷連線，報錯Invocation canceled by client怎麼辦？。
如果Tablestore資料表中有新增的資料，但是Tablestore觸發器沒有被觸發，您可以從以下方面進行排查。關於觸發器不能正常觸發的詳細排查方案可參見觸發器不能正常觸發函數執行怎麼辦？。
- 確認資料表是否開啟了Stream功能，具體請參見步驟一：為資料表開啟Stream功能。
- 確認在建立觸發器時配置的角色是否正確，您可以使用預設的觸發器角色AliyunTableStoreStreamNotificationRole，具體請參見步驟二：建立Tablestore觸發器。
- 查看是否有函數作業記錄，可以根據日誌確認是否是函數執行失敗。函數執行失敗後，會一直重試直到Tablestore中的日誌資料到期。