How to clean data in Tablestore using Function Compute-Tablestore(Tablestore)-阿里云帮助中心

When writing data to Tablestore, you can use Function Compute to cleanse new data and save the results to a separate data table. You can access both the raw and cleansed data in Tablestore in real time.

Use case

Assume that you write log data, which contains the following three fields, to Tablestore. For easier querying, logs with a level greater than 1 are written to a separate data table named result.

Field	Type	Description
id	integer	The ID of the log.
level	integer	The level of the log. A larger value indicates a higher level.
message	string	The content of the log.

Step 1: Enable Tablestore Stream

A Tablestore trigger relies on Tablestore Stream to capture incremental data. Enable Stream on your target table before creating the trigger.

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

Log on to the Tablestore console. In the top navigation bar, select the region where your instance is located, and then click the alias of the target instance.
On the Instance Details page, click the name of the target data table in the Tables list.
Click the Tunnels tab, and then click Enable to the right of Stream Information.
In the Enable Stream dialog box, set the log expiration time, and then click Enable.

The log expiration time must be a non-zero integer in hours, up to a maximum of 168. This setting cannot be changed once configured. Set this value with caution.

Step 2: Create a function and a Tablestore trigger

Create a function.

Log on to the Function Compute console.
Optional: In the upper-right corner of the page, click Go to Function Compute 3.0.
Note
- Function Compute 3.0 offers several enhanced features. This topic uses Function Compute 3.0.
- If you see Back to Function Compute 2.0 in the upper-right corner, you are already using the new console and can skip this step.
In the navigation pane on the left, click Functions.
In the top navigation bar, select a region. On the Functions page, click Create Function.

On the Create Function page, select a creation method, configure the following parameters, and then click Create.

This topic uses an Event Function to demonstrate how to process data changes from Tablestore in real time.

Note

You can use an Event Function, Web Function, or Task Function to process data in Tablestore. For more information, see Technology selection guide.

To automatically trigger data processing based on data changes in Tablestore, create an Event Function. For more information, see Create an event-triggered function.
To trigger data processing with an HTTP request, create a Web Function. For more information, see Create a web function.
To trigger data processing on a schedule or asynchronously, create a Task Function. For more information, see Create a task function.

Basic Settings: Set the Function Name.

Code: Configure the function's runtime and code.

Parameter

Description

Example

Runtime

Select a runtime, such as Python, Java, PHP, Node.js, or a custom container image.

Custom container image.

This topic uses Python 3.9.

Code Upload Method

Select a method to upload your code to Function Compute.

Use Sample Code: The default method. You can select sample code provided by Function Compute to create a function that meets your requirements.
Upload ZIP: Select and upload a ZIP package that contains your function code.
Upload Folder: Select and upload the folder that contains your function code.
OSS: Upload your function code from an OSS bucket by specifying the Bucket Name and Object Name.

Select Use Sample Code and then choose the Hello, world! sample.

Advanced Settings: Configure instance details and the function's execution timeout.

Parameter	Description	Example
Specifications	Select a suitable combination of vCPU Capacity and Memory. For more information about billing, see Billing overview. Note The ratio of vCPUs (in cores) to memory (in GB) must be between 1:1 and 1:4.	0.35 Cores, 512 MB
Size of Temporary Disk	Select a temporary disk size for storing files. Valid values: 512 MB: The default value. You are not charged for this amount, as Function Compute provides a free quota of up to 512 MB. 10 GB: You are charged for 9.5 GB. Note All directories on the temporary disk are writable and share the same disk space. The lifecycle of the temporary disk matches the underlying function instance. Data on the disk is deleted when the instance is reclaimed. To persist files, you can mount NAS or OSS. For more information, see Configure a NAS file system and Configure an OSS mount.	512 MB
Execution Timeout Period	Set the timeout. The default value for Execution Timeout Period is 180 seconds, and the maximum value is 86,400 seconds.	180
Handler	The method that the Function Compute runtime loads and calls to process requests. This parameter is not required if you create a Web Function. Note When you select Use Sample Code for Code Upload Method, you do not need to modify the Handler. If you select another code upload method, you need to modify the Handler based on your specific needs. Otherwise, an error will occur when the function is executed.	index.handler
Time zone	Select the function's time zone. This automatically adds an environment variable TZ set to the selected time zone.	UTC
Function Role	Function Compute uses this RAM role to generate temporary credentials for accessing your Alibaba Cloud resources and passes them to your code. Important The function role must be granted permissions to access Tablestore. For more information, see Use Function Compute.	AliyunFCDefaultRole
Access to VPC	Specify whether to allow the function to access resources in a VPC. For more information, see Configure network access.	Yes
VPC	Required if Access to VPC is set to Yes. Create a new VPC or select the ID of an existing VPC from the drop-down list.	fc.auto.create.vpc.1632317****
vSwitch	Required if Access to VPC is set to Yes. Create a new vSwitch or select a vSwitch ID from the drop-down list.	fc.auto.create.vswitch.vpc-bp1p8248****
Security Group	Required if Access to VPC is set to Yes. Create a new security group or select one from the drop-down list.	fc.auto.create.SecurityGroup.vsw-bp15ftbbbbd****
Allow function to access the public network	Specifies whether the function can access the public network through its default network interface controller (NIC). If you disable this option, the function cannot access the public network. Important If you use a static public IP address, you must disable Allow function to access the public network. Otherwise, the static public IP address does not take effect. For more information, see Configure a static public IP address.	Yes
Logging	Enables or disables Log Service. Valid values: Enable: Function execution logs are stored in Log Service, which helps with debugging, failure analysis, and data analysis. Note When logging is enabled, Log Service collects the function's output to stdout. You can then view the execution logs for debugging, analysis, and more. Click Configure logs for more details. The Log Service resources created by Function Compute incur fees. For more information, see Pay-as-you-go billable items. Disable: Function execution logs cannot be stored or queried through Log Service.	Enable

(Optional) Environment Variables: Configure environment variables for the function's runtime. For more information, see Configure environment variables.

Create a Tablestore trigger.

On the Function Details page, click the Configure tab. In the navigation pane on the left, click Trigger, and then click Create Trigger.

In the Create Trigger panel, configure the parameters and then click OK.

Parameter	Actions	Example value
Trigger Type	Select Tablestore.	Tablestore
Name	Enter a custom name for the trigger.	Tablestore-trigger
Version or Alias	The default value is LATEST. To create a trigger for a different version or alias, you must first select it from the Version or Alias drop-down list on the function details page. For more information, see Manage versions and Manage aliases.	LATEST
Instances	Select an existing Tablestore instance from the list.	d00dd8xm****
Table	Select an existing table from the list.	mytable
Role Name	Select AliyunTableStoreStreamNotificationRole. Note If this is the first time you are creating this type of trigger, click OK, and then click Authorize Now in the dialog box that appears.	AliyunTableStoreStreamNotificationRole

After the trigger is created, it is displayed on the Triggers tab. To modify or delete a trigger, see Trigger Management.

Step 3: Test and verify

After you create the trigger, write data to the Tablestore data table and query the data to verify that the function cleanses data as expected.

On the Function Details page, click the Code tab and enter the following code in the code editor.

This example uses Python. You must replace the values for INSTANCE_NAME, REGION, ENDPOINT, and RESULT_TABLENAME with your actual values.

#!/usr/bin/env python
# -*- coding: utf-8 -*-
import cbor
import json
import tablestore as ots

INSTANCE_NAME = 'distribute-test'
REGION = 'cn-shanghai'
ENDPOINT = 'http://%s.%s.vpc.tablestore.aliyuncs.com' % (INSTANCE_NAME, REGION)
RESULT_TABLENAME = 'result'


def get_attrbute_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']


# The credentials acquired from the context grant access to Tablestore because the function role has the required permissions.
def get_ots_client(context):
    creds = context.credentials
    client = ots.OTSClient(ENDPOINT, creds.access_key_id, creds.access_key_secret, INSTANCE_NAME,
                           sts_token=creds.security_token)
    return client


def save_to_ots(client, record):
    id = int(get_pk_value(record, 'id'))
    level = int(get_attrbute_value(record, 'level'))
    msg = get_attrbute_value(record, 'message')
    pk = [('id', id), ]
    attr = [('level', level), ('message', msg), ]
    row = ots.Row(pk, attr)
    client.put_row(RESULT_TABLENAME, row)


def handler(event, context):
    records = cbor.loads(event)
    # records = json.loads(event)
    client = get_ots_client(context)
    for record in records['Records']:
        level = int(get_attrbute_value(record, 'level'))
        if level > 1:
            save_to_ots(client, record)
        else:
            print("level <= 1, ignore.")

Write data to the source_data data table. Enter values for id, level, and message, and then query the result table to check the cleansed data.
- When you write data with a level greater than 1 to the source_data table, the data is synchronized to the result table.
- When you write data with a level of 1 or less to the source_data table, the data is not synchronized to the result table.

FAQ

If you cannot create a Tablestore trigger in a specific region, ensure the region supports this feature. For more information, see Use Function Compute.
If you cannot find your existing Tablestore data table when creating a trigger, make sure the data table and your function are in the same region.
If you receive a client cancellation error when using a Tablestore trigger, it usually means the client-side timeout is shorter than the function's execution time. Increase the client timeout period. For more information, see What do I do if I receive an 'Invocation canceled by client' error?.
If new data is added to the Tablestore data table but the trigger does not fire, check the following items. For a detailed troubleshooting guide, see What do I do if a trigger fails to execute a function?.
- Confirm that the Stream feature is enabled for the data table. For more information, see Enable the Stream feature for a data table.
- Confirm that the role you configured when you created the trigger is correct. You can use the default trigger role AliyunTableStoreStreamNotificationRole. For more information, see Create a Tablestore trigger.
- Check the function execution logs to see if the function failed. If a function fails, it retries until the stream data in Tablestore expires.
If the function execution fails with an "access_key_id is None or empty." error, confirm that the function's role has the required permissions to access Tablestore. For more information, see Use Function Compute.