使用函数计算做数据清洗_函数触发器_计算与分析

表格存储高并发的写入性能以及低廉的存储成本非常适合物联网、日志、监控数据的存储，您可以将数据写入到表格存储中，同时在函数计算中对新增的数据做简单的清洗，并将清洗之后的数据写回到表格存储表中，并对原始及结果数据提供实时访问。

数据定义

我们假设写入的为日志数据，包括三个基础字段：


字段名称	类型	含义
id	整型	日志id
level	整型	日志的等级，越大表明等级越高
message	字符串	日志的内容

我们需要将 level>1 的日志写入到另外一张数据表中，用作专门的查询。

创建实例及数据表

在表格存储控制台创建表格存储实例（本次以华东2 distribute-test 为例），并创建源表（source_data）及结果表（result），主键为 id（整型），由于表格存储是 schemafree 结构，无需预先定义其他属性列字段。

以 source_data 为例，创建如下图：

开启数据源表的Stream功能

触发器功能需要先开启数据表的Stream功能，才能在函数计算中处理写入表格存储中的增量数据。

Stream记录过期时长是指通过 StreamAPI 能够读取到的增量数据的最长时间。

由于触发器只能绑定现有的函数，故先到函数计算的控制台上在同region创建服务及函数。

创建函数计算服务

在函数计算的控制台上创建服务及处理函数，我们继续使用华东2节点。

在华东2节点创建服务。
创建函数依次选择空白函数 > 不创建触发器。
- 函数名称为：etl_test，选择 python2.7 环境，在线编辑代码
- 函数入口为：etl_test.handler
- 代码稍后编辑，点击下一步。
进行服务授权。

由于函数计算需要将运行中的日志写入到日志服务中，同时，需要对表格存储的表进行读写，故需要对函数计算进行授权，为方便起见，我们先添加 AliyunOTSFullAccess 与 AliyunLogFullAccess 权限，实际生产中，建议根据权限最小原则来添加权限。
单击授权完成，并创建函数。

修改函数代码。

创建好函数之后，点击对应的函数—代码执行，编辑代码并保存，其中，INSTANCE_NAME（表格存储的实例名称）、REGION（使用的区域）需要根据情况进行修改：

使用示例代码如下：

#!/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 _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']
#由于已经授权了AliyunOTSFullAccess权限，此处获取的credentials具有访问表格存储的权限
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."

绑定触发器

回到表格存储的实例管理页面，单击表 source_data 后的使用触发器按钮，进入触发器绑定界面，单击使用已有函数计算，选择刚创建的服务及函数，勾选表格存储发送事件通知的权限，进行确定。
绑定成功之后，能够看到如下的信息：

运行验证

向 source_data 表中写入数据。

单击source_data的数据管理页面，然后单击插入数据，如图依次填入id、level及message信息。
在 result 表中查询清洗后的数据。

单击result表的数据管理页面，会查询到刚写入到 source_data 中的数据。

向 soure_data 写入level <=1的数据将不会同步到 result 表中。