數據同步前準備：Tablestore Stream環境準備

SyncClient client = new SyncClient("", "", "", "");
#方法1：建表的時候開啟：
CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta);
createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // 24代表增量數據保留24小時。
client.createTable(createTableRequest);
#方法2：如果建表時未開啟，您可以通過UpdateTable開啟:
UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName");
updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24));
client.updateTable(updateTableRequest);

支持的同步模式與字段類型

Tablestore Stream Reader插件支持使用列模式或行模式同步Tablestore的增量數據。兩種模式下的同步過程和字段類型要求如下。

列模式

在Tablestore多版本模式下，表中的數據組織為行>列>版本三級的模式， 一行可以有任意列，列名并不是固定的，每一列可以含有多個版本，每個版本都有一個特定的時間戳（版本號）。

您可以通過Tablestore的API進行一系列讀寫操作，Tablestore通過記錄您最近對表的一系列寫操作（或數據更改操作）來實現記錄增量數據的目的，所以您也可以把增量數據看作一批操作記錄。

同時，由于Tablestore具有動態列、多版本的特性，所以Reader插件導出的一行不對應Tablestore中的一行，而是對應Tablestore中的一列的一個版本。即Tablestore中的一行可能會導出很多行，每行包含主鍵值、該列的列名、該列下該版本的時間戳（版本號）、該版本的值、操作類型。如果設置isExportSequenceInfo為true，還會包括時序信息。

行模式

• 寬行表

  您可以通過行模式導出數據，該模式將用戶每次更新的記錄，抽取成行的形式導出，需要設置mode屬性并配置列名。

  "parameter": {
    #parameter中配置下面三項配置（例如datasource、table等其它配置項照常配置）。
    "mode": "single_version_and_update_only", # 配置導出模式。
    "column":[  #按照需求添加需要導出TableStore中的列，您可以自定義設置配置個數。
            {
               "name": "uid"  #列名示例，可以是主鍵或屬性列。
            },
            {
               "name": "name"  #列名示例，可以是主鍵或屬性列。
            },
    ],
    "isExportSequenceInfo": false, #single_version_and_update_only模式下只能是false。
  }

• 時序表

  時序表在創建時會自動開啟Stream，因此不需要手動開啟Stream功能。

  Tablestore Stream Reader支持導出時序表中的增量數據，當表為時序表時，需要配置的信息如下：

  "parameter": {
    #parameter中配置下面四項配置（例如datasource、table等其它配置項照常配置）。
    "mode": "single_version_and_update_only", # 配置導出模式。
    "isTimeseriesTable":"true",  # 配置導出為時序表。
    "column":[  #按照需求添加需要導出TableStore中的列，您可以自定義設置配置個數。
            {
              "name": "_m_name"       #度量名稱字段。
            },
            {
              "name": "_data_source"  #數據源字段。
            },
            {
              "name": "_tags"         #標簽字段，將tags轉換為string類型。
            },
            {
              "name": "tag1",       #標簽內部字段鍵名稱。
              "is_timeseries_tag":"true"  #表明該字段為tags內部字段。
            },
            {
              "name": "time"          #時間戳字段。
            },
            {
               "name": "name"         #屬性列名稱。
            },
    ],
    "isExportSequenceInfo": false, #single_version_and_update_only模式下只能是false。
  }

  行模式導出的數據更接近于原始的行，易于后續處理，但需要注意以下問題：

  • 每次導出的行是從用戶每次更新的記錄中抽取，每一行數據與用戶的寫入或更新操作一一對應。如果用戶存在單獨更新某些列的行為，則會出現有一些記錄只有被更新的部分列，其它列為空的情況。

  • 行模式不會導出數據的版本號（即每列的時間戳），也無法進行刪除操作。

數據類型轉換列表

數據同步任務開發：Tablestore增量同步流程引導

• 操作流程請參見通過向導模式配置離線同步任務、通過腳本模式配置離線同步任務。

• 腳本模式配置的全量參數和腳本Demo請參見下文的附錄：Tablestore Stream腳本Demo與參數說明。

附錄：Tablestore Stream腳本Demo與參數說明

附錄：離線任務腳本配置方式

如果您配置離線任務時使用腳本模式的方式進行配置，您需要在任務腳本中按照腳本的統一格式要求編寫腳本中的reader參數和writer參數，腳本模式的統一要求請參見通過腳本模式配置離線同步任務，以下為您介紹腳本模式下的數據源的Reader參數和Writer參數的指導詳情。

Tablestore Stream Reader腳本Demo

• 列模式

  {
      "type":"job",
      "version":"2.0",//版本號。
      "steps":[
          {
              "stepType":"otsstream",//插件名。
              "parameter":{
                  "datasource":"$srcDatasource",//數據源。
                  "dataTable":"",//表名。
                  "statusTable":"TableStoreStreamReaderStatusTable",//用于記錄狀態的表的名稱。
                  "maxRetries":30,//從 TableStore 中讀增量數據時，每次請求的最大重試次數，默認為30。
                  "isExportSequenceInfo":false,//是否導出時序信息。
                  "startTimeString":"${startTime}${hh}",//增量數據的時間范圍（左閉右開）的左邊界。參數配置中配置startTime=${yyyymmdd} hh=$[hh24miss],表示ots讀取開始時間為業務日期的定時時間
                  "endTimeString":"${endTime}${hh}"http://增量數據的時間范圍（左閉右開）的右邊界。參數配置中配置endTime=${yyyymmdd} hh=$[hh24miss]，表示ots讀取結束時間為業務日期的定時時間
              },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"http://錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業并發數。
              "mbps":"12"http://限流，此處1mbps = 1MB/s。
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• 行模式讀取寬表

  {
      "type":"job",
      "version":"2.0",//版本號。
      "steps":[
          {
              "stepType":"otsstream",//插件名。
              "parameter":{
                  "datasource":"$srcDatasource",//數據源。
                  "dataTable":"",//表名。
                  "statusTable":"TableStoreStreamReaderStatusTable",//用于記錄狀態的表的名稱。
                  "maxRetries":30,//從 TableStore 中讀增量數據時，每次請求的最大重試次數，默認為30。
                  "isExportSequenceInfo":false,//是否導出時序信息。
                  "startTimeString":"${startTime}${hh}",//增量數據的時間范圍（左閉右開）的左邊界。參數配置中配置startTime=${yyyymmdd} hh=$[hh24miss],表示ots讀取開始時間為業務日期的定時時間
                  "endTimeString":"${endTime}${hh}",//增量數據的時間范圍（左閉右開）的右邊界。參數配置中配置endTime=${yyyymmdd} hh=$[hh24miss]，表示ots讀取結束時間為業務日期的定時時間
                  "mode": "single_version_and_update_only",
                  "column":[
                          {
                              "name":"pId"
                          },
                          {
                              "name": "uId"
                          },
                          {
                              "name":"col0"
                          },
                          {
                              "name": "col1"
                          }
                      ],
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"http://錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業并發數。
              "mbps":"12"http://限流
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• 行模式讀取時序表

  {
      "type":"job",
      "version":"2.0",//版本號。
      "steps":[
          {
              "stepType":"otsstream",//插件名。
              "parameter":{
                  "datasource":"$srcDatasource",//數據源。
                  "dataTable":"",//表名。
                  "statusTable":"TableStoreStreamReaderStatusTable",//用于記錄狀態的表的名稱。
                  "maxRetries":30,//從 TableStore 中讀增量數據時，每次請求的最大重試次數，默認為30。
                  "isExportSequenceInfo":false,//是否導出時序信息。
                  "startTimeString":"${startTime}${hh}",//增量數據的時間范圍（左閉右開）的左邊界。參數配置中配置startTime=${yyyymmdd} hh=$[hh24miss],表示ots讀取開始時間為業務日期的定時時間
                  "endTimeString":"${endTime}${hh}",//增量數據的時間范圍（左閉右開）的右邊界。參數配置中配置endTime=${yyyymmdd} hh=$[hh24miss]，表示ots讀取結束時間為業務日期的定時時間
                  "mode": "single_version_and_update_only",
                  "isTimeseriesTable":"true",
                  "column": [
                            {
                              "name": "_m_name"
                            },
                            {
                              "name": "_data_source",
                            },
                            {
                              "name": "_tags",
                            },
                            {
                              "name": "string_column",
                            }
                      ]
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"http://錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業并發數。
              "mbps":"12"http://限流，此處1mbps = 1MB/s。
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

Tablestore Stream Reader腳本參數

"column":[
    {"name":"pk1"},
	{"name":"col1"},
	{"name":"col2","dataType":"new"},
	{"name":"col2","dataType":"old"},
	{"name":"col2","dataType":"latest"}
],