Function Compute:Tablestore triggers

How it works

1. Enable the Stream feature on a Tablestore data table to capture row-level changes as a stream of incremental records.

2. Create a Tablestore trigger in Function Compute and bind it to the table.

3. When a row changes, Tablestore encodes the incremental data in Concise Binary Object Representation (CBOR) format and passes it as the event parameter to your function.

4. Your function parses the event, inspects the operation type (PutRow, UpdateRow, or DeleteRow), and processes the record accordingly.

Limits

• Tablestore triggers are supported in the following regions: China (Beijing), China (Hangzhou), China (Shanghai), China (Shenzhen), China (Hong Kong), Japan (Tokyo), Singapore, and Germany (Frankfurt).

• The Tablestore data table and the Function Compute service must be in the same region.

• To access a function associated with a Tablestore trigger over an internal network, use the VPC endpoint format {instance}.{region}.vpc.tablestore.aliyuncs.com. Do not use the Tablestore internal endpoint in this case.

• Function execution duration cannot exceed one minute.

Usage notes

• Avoid invocation loops. If your function writes back to the same table that triggers it, the write creates another change event, which triggers the function again. Design your data flow so that function output goes to a different table.

• Retry behavior. When a function invocation fails, the trigger retries indefinitely until the Stream log data expires. The log expiration period is set when you enable the Stream feature and cannot be changed afterward.

Prerequisites

Before you begin, make sure you have:

• A Function Compute service and function. See Create a service and Create a function

• A Tablestore instance and data table in the same region as your Function Compute service. See Create instances and Create tables

Step 1: Enable the Stream feature for the data table

The Stream feature captures row-level changes to a Tablestore table as incremental data. Enable it before creating the trigger.

1. Log on to the Tablestore console.

2. In the top navigation bar, select the region.

3. On the Overview page, click the instance name or click Manage Instance in the Actions column.

4. On the Instance Details tab, click the Tables tab. Click the name of the data table, then click the Tunnels tab. Alternatively, click the icon and select Tunnels.

5. On the Tunnels tab, click Enable in the Stream Information section.

6. In the Enable Stream dialog box, set the Log Expiration Time and click Enable. The value must be a non-zero integer in hours. Maximum: 168 hours.

Step 2: Create a Tablestore trigger

1. Log on to the Function Compute console.

2. In the left-side navigation pane, click Services & Functions.

3. In the top navigation bar, select the region.

4. On the Services page, find the service and click Functions in the Actions column.

5. Click the function name to open the function details page.

6. On the Triggers tab, select a version or alias from the Version or Alias drop-down list, then click Create Trigger.

7. In the Create Trigger panel, set the following parameters and click OK.

After the trigger is created, it appears on the Triggers tab. To modify or delete a trigger, see Manage triggers.

Step 3: Understand the event format

The trigger encodes incremental data in CBOR format and passes it as the event parameter to your function. The Records[].Type field identifies the operation (PutRow, UpdateRow, or DeleteRow) — use this field to branch your processing logic for inserts, updates, and deletes.

The following examples show events for all three operation types.

PutRow (row inserted):

{
    "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
                }
            ]
        }
    ]
}

UpdateRow (row updated) and DeleteRow (row deleted):

{
    "Version": "Sync-v1",
    "Records": [
        {
            "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": 1506416610000000
            },
            "PrimaryKey": [
                {
                    "ColumnName": "pk_0",
                    "Value": 9999999999999999
                }
            ],
            "Columns": [
                {
                    "Type": "DeleteAllVersions",
                    "ColumnName": "attr_0",
                    "Timestamp": 1506416610000
                }
            ]
        }
    ]
}

Step 4: Write and test the function

import logging
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")
    # The event is JSON-parsed from the CBOR-encoded payload
    records = json.loads(event)
    for record in records['Records']:
        logger.info("Handle record: %s", record)
        operation_type = record['Type']  # PutRow, UpdateRow, or DeleteRow
        pk_0 = get_pk_value(record, "pk_0")
        if operation_type == 'PutRow':
            attr_0 = get_attribute_value(record, "attr_0")
            logger.info("Insert: pk_0=%s, attr_0=%s", pk_0, attr_0)
        elif operation_type == 'UpdateRow':
            attr_0 = get_attribute_value(record, "attr_0")
            logger.info("Update: pk_0=%s, attr_0=%s", pk_0, attr_0)
        elif operation_type == 'DeleteRow':
            logger.info("Delete: pk_0=%s", pk_0)
    return 'OK'

import com.aliyun.fc.runtime.Context;
import com.aliyun.fc.runtime.StreamRequestHandler;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import java.io.*;
import java.nio.charset.StandardCharsets;

public class TablestoreTriggerHandler implements StreamRequestHandler {
    @Override
    public void handleRequest(InputStream inputStream, OutputStream outputStream, Context context)
            throws IOException {
        String event = new String(inputStream.readAllBytes(), StandardCharsets.UTF_8);
        JsonObject payload = new Gson().fromJson(event, JsonObject.class);
        JsonArray records = payload.getAsJsonArray("Records");
        for (int i = 0; i < records.size(); i++) {
            JsonObject record = records.get(i).getAsJsonObject();
            String type = record.get("Type").getAsString();
            context.getLogger().info("Operation type: " + type);
            // Add your per-operation logic here: PutRow, UpdateRow, or DeleteRow
        }
        outputStream.write("OK".getBytes(StandardCharsets.UTF_8));
    }
}

const handler = async (event, context) => {
    const payload = JSON.parse(event.toString());
    for (const record of payload.Records) {
        const type = record.Type; // PutRow, UpdateRow, or DeleteRow
        context.logger.info(`Operation type: ${type}`);
        if (type === 'PutRow') {
            // Handle insert
        } else if (type === 'UpdateRow') {
            // Handle update
        } else if (type === 'DeleteRow') {
            // Handle delete
        }
    }
    return 'OK';
};

module.exports.handler = handler;

Troubleshooting

The trigger is not created in a specific region. Check whether the region supports Tablestore triggers. For the supported region list, see Limits.

The Tablestore table does not appear when creating the trigger. The table must be in the same region as your Function Compute service. Verify that both resources are in the same region and try again.

The error "Invocation canceled by client" appears repeatedly. The client-side timeout is shorter than the function execution duration. Increase the client timeout. For details, see What do I do if the client is disconnected and the message "Invocation canceled by client" is reported?

Data is written to the table but the trigger does not fire. Work through the following checks:

1. Verify that the Stream feature is enabled on the table. See Step 1: Enable the Stream feature for the data table.

2. Verify that the trigger was created with the correct role (AliyunTableStoreStreamNotificationRole). See Step 2: Create a Tablestore trigger.

3. Check the function run logs to see whether the function is executing but failing. If the function fails, it retries until the Stream log data expires.

What's next

• Manage triggers — modify or delete existing triggers

• Manage versions — manage function versions and aliases