DataWorks：Tablestore Stream資料來源

資料同步前準備：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}"//增量資料的時間範圍（左閉右開）的右邊界。參數配置中配置endTime=${yyyymmdd} hh=$[hh24miss]，表示ots讀取結束時間為業務日期的定時時間
              },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"//錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業並發數。
              "mbps":"12"//限流，此處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"//錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業並發數。
              "mbps":"12"//限流
  
          }
      },
      "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"//錯誤記錄數。
          },
          "speed":{
              "throttle":true,//當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,表示限流。
              "concurrent":1,//作業並發數。
              "mbps":"12"//限流，此處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"}
],