使用函数计算

使用场景

如上图所示，您可以使用函数计算实现以下任务。

• 数据同步：同步表格存储中的实时数据到数据缓存、搜索引擎或者其他数据库实例中。
• 数据归档：增量归档表格存储中的数据到 OSS 等做冷备份。
• 事件驱动：利用触发器触发函数调用 IOT 套件、云应用 API 或者做消息通知等。

配置Tablestore触发器

您可以通过控制台创建Tablestore触发器来处理Tablestore数据表的实时数据流。

1. 创建开通了Stream流的数据表。
   1. 在表格存储控制台创建实例。
   2. 在该实例下创建数据表，并且开启数据表的Stream功能。
2. 创建函数计算的函数。
   1. 登录函数计算控制台。
   2. 在左侧导航栏，选择服务-函数。
   3. 单击新建服务，输入服务名称。单击提交。
   4. 在服务-函数页面，单击新创建的服务。单击服务配置页签，选择角色配置可配置服务的角色，用于授权函数进行日志收集，以及在计算函数中继续访问用户的其他资源。详细信息请参见函数计算权限模型。
   5. 在新创建的服务下单击新建函数。
   6. 在函数模板页面选择 空白函数。
   7. 在触发器配置页面选择不创建任何触发器 ，然后单击下一步。
   8. 配置函数信息。

      Tablestore触发器会使用CBOR格式将增量数据编码为函数计算的Event，并调用用户函数。下图示例函数就是将编码后的Event重新解码和打印到日志中心，您可以在解码数据后进行任意数据处理。

      
3. 创建和测试Tablestore触发器。
   1. 在表格存储管理控制台新创建的数据表下，选择使用已有函数计算来创建触发器。
   2. 创建过程中需要授权Tablestore发送事件通知所需的权限。

      勾选完毕后，实际可以在RAM控制台查看到自动创建的授权角色AliyunTableStoreStreamNotificationRole。

数据处理

• 数据格式

  Tablestore触发器使用CBOR格式对增量数据进行编码构成函数计算的Event。增量数据的具体数据格式如下：

  {
      "Version": "string",
      "Records": [
          {
              "Type": "string",
              "Info": {
                  "Timestamp": int64
              },
              "PrimaryKey": [
                  {
                      "ColumnName": "string",
                      "Value": formated_value
                  }
              ],
              "Columns": [
                  {
                      "Type": "string",
                      "ColumnName": "string",
                      "Value": formated_value,
                      "Timestamp": int64
                  }
              ]
          }
      ]
  }

• 成员定义
  • Version
    • 含义：payload版本号，目前为Sync-v1
    • 类型：string
  • Records
    • 含义：数据表中的增量数据行数组
    • 内部成员：
      • Type
        • 含义：数据行类型，包含PutRow、UpdateRow和DeleteRow
        • 类型： string
      • Info
        • 含义：数据行基本信息
        • 内部成员：
          • Timestamp
          • 含义： 该行的最后修改UTC时间
          • 类型：int64
      • PrimaryKey
        • 含义：主键列数组
        • 内部成员
          • ColumnName
            • 含义：主键列名称
            • 类型：string
          • Value
            • 含义：主键列内容
            • 类型：formated_value，支持integer、string和blob
      • Colunms
        • 含义：属性列数组
        • 内部成员
          • Type
            • 含义： 属性列类型，包含Put、DeleteOneVersion和DeleteAllVersions
            • 类型： string
          • ColumnName
            • 含义：属性列名称
            • 类型：string
          • Value
            • 含义： 属性列内容
            • 类型： formated_value，支持integer、boolean、double、string和blob
          • Timestamp
            • 含义： 属性列最后修改UTC时间
            • 类型： int64
• 数据示例

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

在线调试

函数计算支持函数的在线调试功能，用户能够自己构建触发的Event，并测试代码逻辑是否符合期望。

由于Tablestore触发函数服务的Event是CBOR格式，该数据格式是一种类似JSON的二进制格式，可以按照如下方法进行在线调试：

1. 在代码中同时import CBOR和JSON。
2. 单击触发事件，选择自定义，将上述数据示例的JSON文件粘贴至编辑框，根据实际情况修改，并保存。
3. 代码中先使用 records = json.loads(event) 来处自定义的测试触发事件，单击执行，查看是否符合期望。
4. 当使用 records=json.loads(event) 测试OK之后，可以修改为 records = cbor.loads(event) 并保存，当Tablestore中有数据写入时，相关的函数逻辑就会触发。
   

import logging
import cbor
import json
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']
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_attrbute_value(record, "attr_0")
    return 'OK'