Use Function Compute for data cleansing

Data definition

Assume that the data to be written is log data, which contains three fields:

Logs of level>1 must be written to another data table for dedicated querying.

Creating instances and data tables

Create a Table Store instance in the Table Store console(this time using East China Node 2 distribute-test as an example), and create a source table(source_data)and a result table(result). The primary key is id (integer). Since Table Store uses a schemafree structure, other attribute column fields need not be predefined.

Taking source_data as an example, create as in the following diagram:

Enabling the streaming function of the data source table

The trigger function requires that the Stream function of the data table be enabled before the incremental data written in Table Store can be processed in Function Compute.

Expiration time of Stream records is the maximum time for incremental data to be read through the stream API.

Since the trigger can bind only existing functions, first create services and functions in the same region on the Function Compute console.

Create a Function Compute service

The following process describes creating services and processing functions on the Function Compute console, using East China Node 2 as an example.

1. Create a service in East China Node 2.
   
   
2. Create a function and select: Blank function > Do not create a trigger.
   
   
   • The function name is: etl_test. Select the Python 2.7 environment, and edit code online
   • The function entry is: etl_test.handler
   • The code is edited later. Now click Next.
3. Service authorization

   Since Function Compute writes the running log to the log service and simultaneously reads and writes the Table Store data tables, Function Compute must have certain permissions. For convenience, first add AliyunOTSFullAccess and AliyunLogFullAccess permissions. In actual usage, we recommend that you add permissions based on the principle of least privilege.

   
   
   
4. Click Complete authorization and create a function.
5. Modify the function code.

   After creating the function, click the corresponding Function—Run Code, then edit the code and save it. Modify INSTANCE_NAME (the Table Store instance name) and REGION (region used) as appropriate:

   
   
   
   Example code:

   #! /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.ots-internal.aliyuncs.com'%(INSTANCE_NAME, REGION)
   RESULT_TABLENAME = 'result'
   def _utf8(input):
       return str(bytearray(input, "utf-8"))
   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']
   #Because the AliyunOTSFullAccess permission has been granted, the credentials obtained here are authorized to access Table Store.
   def get_ots_client(context):
       creds = context.credentials
       client = ots.OTSClient(ENDPOINT, creds.accessKeyId, creds.accessKeySecret, INSTANCE_NAME, sts_token = creds.securityToken)
       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 = [(_utf8('id'), id),]
       attr = [(_utf8('level'), level), (_utf8('message'), _utf8(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."

Binding a Trigger

Run verification