Tablestore：同步增量資料到OSS

（推薦）嚮導模式

1. 在配置任務步驟的配置資料來源與去向地區，根據實際配置資料來源和資料去向。

   資料來源配置

   參數	說明
   表	Tablestore中的資料表名稱。
   開始時間	增量讀取資料的開始時間和結束時間，分別配置為變數形式${startTime}和${endTime}，具體格式在後續調度屬性中配置。增量資料的時間範圍為左閉右開的區間。
   結束時間
   狀態表	用於選項組的表名稱，預設值為TableStoreStreamReaderStatusTable。
   最大重試次數	從TableStore中讀取增量資料時，每次請求的最大重試次數。
   匯出時序資訊	是否匯出時序資訊，時序資訊包含了資料的寫入時間等。

   資料去向配置

   參數	說明
   文本類型	寫入OSS的檔案類型，例如csv、txt。 說明 不同檔案類型支援的配置有差異，請以實際介面為準。
   檔案名稱(含路徑)	當設定文本類型為text、csv或orc時才能配置該參數。 OSS中的檔案名稱，支援帶有路徑，例如tablestore/20231130/myotsdata.csv。
   檔案路徑	當設定文本類型為parquet時才能配置該參數。 檔案在OSS上的路徑，例如tablestore/20231130/。
   檔案名稱	當設定文本類型為parquet時才能配置該參數。 OSS中的檔案名稱。
   資料行分隔符號	當設定文本類型為text或csv時才能配置該參數。 寫入OSS檔案時，列之間使用的分隔字元。
   行分隔字元	當設定文本類型為text時才能配置才參數。 自訂的行分隔字元，用來分隔不同資料行。例如配置為 \u0001。您需要使用資料中不存在的分隔字元作為行分隔字元。 如果要使用Linux或Windows平台的預設行分隔字元（\n、\r\n）建議置空此配置，平台能自適應讀取。
   編碼	當設定文本類型為text或csv時才能配置該參數。 寫入檔案的編碼配置。
   null值	當設定文本類型為text、csv或orc時才能配置該參數。 來源資料源中可以表示為null的字串，例如配置為null，如果來源資料是null，則系統會視作null欄位。
   時間格式	當設定文本類型為text或csv時才能配置該參數。 日期類型的資料寫入到OSS檔案時的時間格式，例如yyyy-MM-dd。
   首碼衝突	當設定的檔案名稱與OSS中已有檔案名稱衝突時的處理方法。取值範圍如下： 替換：刪除原始檔案，重建一個同名檔案。 保留：保留原始檔案，重建一個新檔案，名稱為原檔案名稱加隨機尾碼。 報錯：同步任務停止執行。
   切分檔案	當設定文本類型為text或csv時才能配置該參數。 寫入OSS檔案時，單個Object檔案的最大大小。單位為MB。最大值為100 GB。當檔案大小超過指定的切分檔案大小時，系統會產生新檔案繼續寫入資料，直到完成所有資料寫入。
   寫為一個檔案	當設定文本類型為text或csv時才能配置該參數。 寫入資料到OSS時，是否寫單個檔案。預設寫多個檔案，當讀不到任何資料時，如果配置了檔案頭，則會輸出只包含檔案頭的空檔案，否則只輸出空檔案。 如果需要寫入到單個檔案，請選中寫為一個檔案複選框。此時當讀不到任何資料時， 不會產生空檔案。
   首行輸出表頭	當設定文本類型為text或csv時才能配置該參數。 寫入檔案時第一行是否輸出表頭。預設不輸出表頭。如果需要在第一行輸出表頭，請選中首行輸出表頭複選框。

2. 在欄位對應地區，系統自動進列欄位映射，保持預設配置即可。

   來源欄位中包括了表主鍵和增量變更資訊，目標欄位不支援配置。

   

3. 在通道控制地區，配置任務運行參數，例如任務期望最大並發數、同步速率、髒資料同步策略、分散式處理能力等。關於參數配置的更多資訊，請參見配置通道。

4. 單擊表徵圖，儲存配置。

   說明

   執行後續操作時，如果未儲存配置，則系統會出現儲存確認的提示，單擊確認即可。

指令碼模式

增量資料的同步需要使用到OTSStream Reader和OSS Writer外掛程式。指令碼配置規則請參見Tablestore Stream資料來源和OSS資料來源。

1. 在配置任務步驟，單擊表徵圖，然後在彈出的對話方塊中單擊確認。

2. 在指令碼配置頁面，請根據如下樣本完成配置。

   重要

   為了便於理解，在配置樣本中增加了注釋內容，實際使用指令碼時請刪除所有注釋內容。

   按照行模式匯出增量變化後的資料

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", //外掛程式名，不能修改。
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", //儲存Tablestore Stream狀態的表，一般無需修改。
                   "maxRetries": 30, //最大重試次數。
                   "isExportSequenceInfo": false, //是否匯出時序資訊，時序資訊包含了資料的寫入時間等。
   				        "mode": "single_version_and_update_only", //Tablestore Stream匯出資料的格式，目前需要設定為single_version_and_update_only。如果配置模板中無此項，則需要增加。
                   "datasource": "otssource", //資料來源名稱，請根據實際填寫。
                   "envType": 1,
                   "column": [ 
   				          {
                        "name": "pk1" 
                     },
                     {
                        "name": "pk2"  
                     },
                     {
                        "name": "col1"  
                     }
                   ],
                   "startTimeString": "${startTime}", //開始匯出的時間點，由於是增量匯出，需要迴圈啟動此任務，則此處每次啟動時的時間都不同，因此需要設定一個變數，例如${startTime}。
                   "table": "mytable", //Tablestore中的資料表名稱。
                   "endTimeString": "${endTime}"  //結束匯出的時間點。此處也需要設定一個變數，例如${endTime}。
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", //Writer外掛程式的名稱，不能修改。
               "parameter": { //此處參數配置只適用於匯出csv和text格式的檔案。如果要匯出為parquet和orc格式，請參照OSS資料來源文檔修改OSSWriter配置。
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", //定義null值的字串標識符方式，可以是Null 字元串。
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", //時間格式。
                   "datasource": "osssource", //OSS資料來源名稱，請根據實際填寫。
                   "envType": 1,
                   "writeSingleObject": true, //是否寫單個檔案。設定此參數為true，表示寫單個檔案，當讀不到任何資料時不會產生空檔案；設定此參數為false，表示寫多個檔案，當讀不到任何資料時，如果設定檔頭會輸出空檔案只包含檔案頭，否則只輸出空檔案。
                   "writeMode": "truncate", //當同名檔案存在時系統進行的操作，可選值包括truncate、append和nonConflict。truncate表示會清理已存在的同名檔案，append表示會增加到已存在的同名檔案內容後面，nonConflict表示當同名檔案存在時會報錯。 
                   "encoding": "UTF-8", //編碼類別型。
                   "fieldDelimiter": ",", //每一列的分隔字元。
                   "fileFormat": "csv", //檔案類型，可選值包括csv、text、parquet和orc格式。
                   "object": "" //備份到OSS的檔案名稱首碼，建議使用"Tablestore執行個體名/表名/date"，例如"instance/table/{date}"。
               },
               "name": "Writer",
               "category": "writer" 
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" //允許出錯的個數。當錯誤超過這個數目的時候同步任務會失敗。
           },
           "locale": "zh",
           "speed": {
               "throttle": false, //當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,需要配置mbps參數，表示限流。
               "concurrent": 3 //作業並發數。
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

   按照行的列值增量變化形式匯出資料

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", //外掛程式名，不能修改。
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", //儲存Tablestore Stream狀態的表，一般無需修改。
                   "maxRetries": 30, //最大重試次數。
                   "isExportSequenceInfo": false,  //是否匯出時序資訊，時序資訊包含了資料的寫入時間等。
                   "datasource": "otssource",
                   "envType": 1,
                   "column": [//設定資料表中需要匯出到OSS中的列，如果配置模板中無此項則需要增加，保持預設配置即可。
                       "pk1", //主鍵列名稱，如果有多個主鍵列則需要全部配置。
                       "pk2", //主鍵列名稱，如果有多個主鍵列則需要全部配置。
                       "colName", //表示有增量變化的屬性列名稱，無需修改。
                       "version", //表示增量變化後列的資料版本號碼，無需修改。格式為64位時間戳記。單位為毫秒。
                       "colValue", //表示增量變化後屬性列值，無需修改。
                       "opType", //表示增量操作類型，無需修改。
                       "sequenceInfo" //表示自增保序sequenceid，無需修改。
                   ],
                   "startTimeString": "${startTime}", //開始匯出的時間點，由於是增量匯出，需要迴圈啟動此任務，則此處每次啟動時的時間都不同，因此需要設定一個變數，例如${startTime}。
                   "table": "mytable", //Tablestore中的資料表名稱。
                   "endTimeString": "${endTime}"  //結束匯出的時間點。此處也需要設定一個變數，例如${endTime}。
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", //Writer外掛程式的名稱，不能修改。
               "parameter": { //此處參數配置只適用於匯出csv和text格式的檔案。如果要匯出為parquet和orc格式，請參照OSS資料來源文檔修改OSSWriter配置。
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", //定義null值的字串標識符方式，可以是Null 字元串。
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", //時間格式。
                   "datasource": "osssource", //OSS資料來源名稱，請根據實際填寫。
                   "envType": 1,
                   "writeSingleObject": true, //是否寫單個檔案。設定此參數為true，表示寫單個檔案，當讀不到任何資料時不會產生空檔案；設定此參數為false，表示寫多個檔案，當讀不到任何資料時，如果設定檔頭會輸出空檔案只包含檔案頭，否則只輸出空檔案。
                   "column": [ //匯出到OSS的列，只需要用序號表示即可，無需修改。
                       "0",
                       "1",
                       "2",
                       "3",
                       "4",
                       "5",
                       "6"
                   ],
                   "writeMode": "truncate", //當同名檔案存在時系統進行的操作，可選值包括truncate、append和nonConflict。truncate表示會清理已存在的同名檔案，append表示會增加到已存在的同名檔案內容後面，nonConflict表示當同名檔案存在時會報錯。 
                   "encoding": "UTF-8", //編碼類別型。
                   "fieldDelimiter": ",", //每一列的分隔字元。
                   "fileFormat": "csv", //檔案類型，可選值包括csv和text格式。
                   "object": "" //備份到OSS的檔案名稱首碼，建議使用"Tablestore執行個體名/表名/date"，例如"instance/table/{date}"。
               },
               "name": "Writer",
               "category": "writer"
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" //允許出錯的個數。當錯誤超過這個數目的時候同步任務會失敗。
           },
           "locale": "zh",
           "speed": {
               "throttle": false, //當throttle值為false時，mbps參數不生效，表示不限流；當throttle值為true時,需要配置mbps參數，表示限流。
               "concurrent": 3 //作業並發數。
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

3. 單擊表徵圖，儲存配置。

   說明

   執行後續操作時，如果未儲存指令碼，則系統會出現儲存確認的提示，單擊確認即可。