Function Compute:Tablestore トリガー

前提条件

開始する前に、以下を準備してください。

• FC の関数。詳細については、「関数の作成」をご参照ください。

• Tablestore インスタンス。詳細については、「インスタンスの作成」をご参照ください。

• そのインスタンス内のテーブル。詳細については、「テーブルの作成」をご参照ください。

制限事項

• Tablestore トリガーは、以下のリージョンでサポートされています：中国 (北京)、中国 (杭州)、中国 (上海)、中国 (深圳)、日本 (東京)、シンガポール、ドイツ (フランクフルト)、香港 (中国)。

• Tablestore テーブルは、関数と同じリージョンに存在する必要があります。

• Tablestore トリガーによって呼び出される関数の実行は、1 分を超えることはできません。

• 内部ネットワーク経由で Tablestore にアクセスするには、{instance}.{region}.vpc.tablestore.aliyuncs.com の形式で Virtual Private Cloud (VPC) エンドポイントを使用します。

注意事項

呼び出しループの回避：関数が、自身をトリガーしたテーブルと同じテーブルに書き戻す場合 (たとえば、テーブル A が関数 B をトリガーし、関数 B がテーブル A を更新する場合)、トリガーが再度起動し、無限ループが発生します。関数の出力が別のテーブルに向かうようにデータフローを設計してください。

失敗時のリトライ動作：関数の実行が失敗した場合、トリガーは Tablestore のログデータが期限切れになるまでリトライを続けます。

ステップ 1：データテーブルのストリーム機能の有効化

Tablestore のストリーム機能は、データの増分変更をキャプチャし、関数に配信します。トリガーを作成する前に、データテーブルでこの機能を有効にしてください。

1. Tablestore コンソール にログインします。

2. 上部メニューで、リージョンを選択します。

3. [概要] ページで、インスタンスのエイリアスをクリックするか、[操作] 列の [インスタンスの管理] をクリックします。

4. [インスタンス詳細] タブで、[データテーブル] タブをクリックします。次に、テーブル名をクリックして [Stream] タブを選択するか、 をクリックして [Stream] を選択します。

5. [Stream] タブで、ストリーム情報の横にある [有効化] をクリックします。

6. [ストリームの有効化] ダイアログボックスで、[ログの有効期限] パラメーターを設定し、[有効化] をクリックします。値には、時間単位でゼロ以外の整数を指定する必要があります。最大値は 168 時間です。

ステップ 2：Tablestore トリガーの作成

1. Function Compute コンソール にログインします。左側メニューで、[関数] をクリックします。

2. 上部メニューでリージョンを選択します。[関数] ページで、設定する関数をクリックします。

3. 関数の詳細ページで、[設定] タブをクリックします。左側メニューで [トリガー] をクリックし、[トリガーの作成] をクリックします。

4. [トリガーの作成] パネルで、以下のパラメーターを設定し、[OK] をクリックします。

トリガーが作成されると、[トリガー] タブに表示されます。トリガーを変更または削除するには、「トリガーの管理」をご参照ください。

ステップ 3：テストパラメーターの設定

Tablestore からデータが到着する前に関数をテストするには、Tablestore トリガーのペイロード形式に一致するテストイベントを設定します。

1. 関数の詳細ページの [コード] タブで、[関数のテスト] の横にある アイコンをクリックし、[テストパラメーターの設定] を選択します。

2. [テストパラメーターの設定] パネルで、[新しいテストイベントの作成] または [既存のテストイベントの変更] をクリックし、イベント名とイベント内容を入力して [OK] をクリックします。

Tablestore トリガーは、増分データを Concise Binary Object Representation (CBOR) 形式でエンコードし、イベントとして関数に渡します。以下の例は、PutRow、UpdateRow、DeleteRow の 3 種類のレコードタイプを持つペイロードを示しており、操作によって構造がどのように異なるかを確認できます：

{
    "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
                }
            ]
        },
        {
            "Type": "UpdateRow",
            "Info": {
                "Timestamp": 1506416600000000
            },
            "PrimaryKey": [
                {
                    "ColumnName": "pk_0",
                    "Value": 1506416585881590900
                }
            ],
            "Columns": [
                {
                    "Type": "Put",
                    "ColumnName": "attr_0",
                    "Value": "updated_value",
                    "Timestamp": 1506416600000
                },
                {
                    "Type": "DeleteOneVersion",
                    "ColumnName": "attr_1",
                    "Timestamp": 1506416585741
                }
            ]
        },
        {
            "Type": "DeleteRow",
            "Info": {
                "Timestamp": 1506416700000000
            },
            "PrimaryKey": [
                {
                    "ColumnName": "pk_0",
                    "Value": 1506416585881590900
                }
            ],
            "Columns": []
        }
    ]
}

以下の表に、イベントフィールドを示します。

ステップ 4：関数コードの記述とテスト

トリガーが作成されたら、Tablestore イベントのペイロードを処理する関数コードを記述します。テーブル内のデータが変更されると、関数は自動的に呼び出されます。

1. 関数の詳細ページで、[コード] タブをクリックし、関数コードを入力して [デプロイ] をクリックします。以下の Python の例では、Tablestore のイベントレコードを読み取り、プライマリキーと属性列の値を抽出します：

   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'

2. [関数のテスト] をクリックします。

実行後、[コード] タブで結果を確認できます。

よくある質問

特定のリージョンでトリガーの作成が失敗するのはなぜですか？

Tablestore トリガーは、特定のリージョンでのみ利用可能です。お使いのリージョンがサポートされているか、「制限事項」セクションでご確認ください。

トリガーを作成するときに Tablestore テーブルが見つからないのはなぜですか？

Tablestore テーブルは、関数と同じリージョンに存在する必要があります。テーブルが異なるリージョンにある場合、トリガーの設定画面には表示されません。

「Invocation canceled by client」というエラーが表示され続けるのはなぜですか？

このエラーは通常、クライアント側のタイムアウトが実際の関数の実行時間よりも短いことを意味します。クライアントのタイムアウト期間を延長してください。詳細については、「クライアントが切断され、「Invocation canceled by client」というメッセージが報告された場合の対処方法」をご参照ください。

テーブルにデータを書き込んだ後、Tablestore トリガーが起動しないのはなぜですか？

以下を確認してください：

1. テーブルでストリーム機能が有効になっていることを確認してください。「ステップ 1：データテーブルのストリーム機能の有効化」をご参照ください。

2. トリガーの作成時に正しいロール (AliyunTableStoreStreamNotificationRole) が設定されていることを確認してください。「ステップ 2：Tablestore トリガーの作成」をご参照ください。

3. 関数の実行ログをチェックして、実行が失敗していないか確認してください。関数が繰り返し失敗する場合、ログデータが期限切れになるまでリトライします。「トリガーが関数の実行をトリガーできない場合の対処方法」をご参照ください。