OTSStream Reader

背景信息

OTSStream Reader插件主要用于导出Table Store的增量数据。您可以将增量数据看作操作日志，除数据本身外还附有操作信息。

与全量导出插件不同，增量导出插件只有多版本模式，且不支持指定列。使用插件前，您必须确保表上已经开启Stream功能。您可以在建表时指定开启，也可以使用SDK的UpdateTable接口开启。

SyncClient client = new SyncClient("", "", "", "");
建表的时候开启：
CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta);
createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // 24代表增量数据保留24小时。
client.createTable(createTableRequest);
如果建表时未开启，您可以通过UpdateTable开启:
UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName");
updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); 
client.updateTable(updateTableRequest);

您使用SDK的UpdateTable功能，指定开启Stream并设置过期时间，即开启了Table Store增量数据导出功能。开启后，Table Store服务端就会将您的操作日志额外保存起来，每个分区有一个有序的操作日志队列，每条操作日志会在一定时间后被垃圾回收，该时间即为您指定的过期时间。

Table Store的SDK提供了几个Stream相关的API用于读取这部分的操作日志，增量插件也是通过Table Store SDK的接口获取到增量数据，默认情况下会将增量数据转化为多个6元组的形式（pk、colName、version、colValue、opType和sequenceInfo）导入至MaxCompute中。

列模式

在Table Store多版本模型下，表中的数据组织为行>列>版本三级的模式， 一行可以有任意列，列名并不是固定的，每一列可以含有多个版本，每个版本都有一个特定的时间戳（版本号）。

您可以通过Table Store的API进行一系列读写操作，Table Store通过记录您最近对表的一系列写操作（或数据更改操作）来实现记录增量数据的目的，所以您也可以把增量数据看作一批操作记录。

Table Store会根据每种操作生成对应的增量数据记录，Reader插件会读出这些记录，并导出为数据集成的数据格式。

同时，由于Table Store具有动态列、多版本的特性，所以Reader插件导出的一行不对应Table Store中的一行，而是对应Table Store中的一列的一个版本。即Table Store中的一行可能会导出很多行，每行包含主键值、该列的列名、该列下该版本的时间戳（版本号）、该版本的值、操作类型。如果设置isExportSequenceInfo为true，还会包括时序信息。

行模式

"parameter": {
  #parameter中配置下面三项配置（例如datasource、table等其它配置项照常配置）。
  "mode": "single_version_and_update_only", # 配置导出模式。
  "column":[  #按照需求添加需要导出TableStore中的列，您可以自定义设置配置个数。
          {
             "name": "uid"  #列名示例，可以是主键或属性列。
          },
          {
             "name": "name"  #列名示例，可以是主键或属性列。
          },
  ],
  "isExportSequenceInfo": false, #single_version_and_update_only模式下只能是false。
}

数据类型转换列表

参数说明

向导开发介绍

1. 选择数据源。
   配置同步任务的数据来源和数据去向。

   参数	描述
   数据源	即上述参数说明中的dataSource，通常选择您配置的数据源名称。
   表	即上述参数说明中的dataTable。
   开始时间	任务的开始时间，即增量数据的时间范围（左闭右开）的左边界。格式为yyyymmddhh24miss，单位为毫秒。 说明 开始时间需要设置为最近七天（包括当天）。
   结束时间	任务的结束时间，即增量数据的时间范围（左闭右开）的右边界。格式为yyyymmddhh24miss，单位为毫秒。
   状态表	用于记录状态的表的名称。
   最大重试次数	即上述参数说明中的maxRetries，默认值为30。
   导出时序信息	即上述参数说明中的isExportSequenceInfo，默认值为false。

2. 字段映射，即上述参数说明中的column。
   左侧的源头表字段和右侧的目标表字段为一一对应的关系。单击添加一行可以增加单个字段，鼠标放至需要删除的字段上，即可单击删除图标进行删除。

   参数	说明
   同名映射	单击同名映射，可以根据名称建立相应的映射关系，请注意匹配数据类型。
   同行映射	单击同行映射，可以在同行建立相应的映射关系，请注意匹配数据类型。
   取消映射	单击取消映射，可以取消建立的映射关系。

3. 通道控制。

   参数	描述
   任务期望最大并发数	数据同步任务内，可以从源并行读取或并行写入数据存储端的最大线程数。向导模式通过界面化配置并发数，指定任务所使用的并行度。
   同步速率	设置同步速率可以保护读取端数据库，以避免抽取速度过大，给源库造成太大的压力。同步速率建议限流，结合源库的配置，请合理配置抽取速率。
   错误记录数	错误记录数，表示脏数据的最大容忍条数。
   分布式处理能力	数据同步时，可以将任务切片分散到多台执行节点上并发执行，提高同步速率。该模式下，配置较大任务并发数会增加数据存储访问压力，如需使用该功能，请提前评估数据存储的访问负载。该功能仅支持在独享数据集成资源组配置，详情请参见独享数据集成资源组和新增和使用独享数据集成资源组。

脚本开发介绍

{
    "type":"job",
    "version":"2.0",//版本号。
    "steps":[
        {
            "stepType":"otsstream",//插件名。
            "parameter":{
                "statusTable":"TableStoreStreamReaderStatusTable",//用于记录状态的表的名称。
                "maxRetries":30,//从 TableStore 中读增量数据时，每次请求的最大重试次数，默认为30。
                "isExportSequenceInfo":false,//是否导出时序信息。
                "datasource":"$srcDatasource",//数据源。
                "startTimeString":"${startTime}${hh}",//增量数据的时间范围（左闭右开）的左边界。参数配置中配置startTime=${yyyymmdd} hh=$[hh24miss],表示ots读取开始时间为业务日期的定时时间
                "table":"",//表名。
                "endTimeString":"${endTime}${hh}"//增量数据的时间范围（左闭右开）的右边界。参数配置中配置endTime=${yyyymmdd} hh=$[hh24miss]，表示ots读取结束时间为业务日期的定时时间
            },
            "name":"Reader",
            "category":"reader"
        },
        { 
            "stepType":"stream",
            "parameter":{},
            "name":"Writer",
            "category":"writer"
        }
    ],
    "setting":{
        "errorLimit":{
            "record":"0"//错误记录数。
        },
        "speed":{
            "throttle":true,//当throttle值为flase时，mbps参数不生效，表示不限流；当throttle值为true时,表示限流。
            "concurrent":1 //作业并发数。
            "mbps":"12"//限流

        }
    },
    "order":{
        "hops":[
            {
                "from":"Reader",
                "to":"Writer"
            }
        ]
    }
}