Tablestore：同步增量資料到OSS

步驟一：新增Table Store資料來源

在DataWorks中配置Table Store資料來源，建立與來源資料的串連。

1. 登入DataWorks控制台，切換至目標地區後，單擊左側導覽列的Data Integration > Data Integration，在下拉框中選擇對應工作空間後單擊進入Data Integration。

2. 在左側導覽列，單擊資料來源。

3. 在資料來源列表頁面，單擊新增資料來源。

4. 在新增資料來源對話方塊，搜尋並選擇資料來源類型為Tablestore。

5. 在新增OTS資料來源對話方塊，根據下表配置資料來源參數。

   參數	說明
   資料來源名稱	資料來源名稱必須以字母、數字、底線（_）組合，且不能以數字和底線（_）開頭。
   資料來源描述	對資料來源進行簡單描述，不得超過80個字元。
   地區	選擇Tablestore執行個體所屬地區。
   Table Store執行個體名稱	Tablestore執行個體的名稱。
   Endpoint	Tablestore執行個體的服務地址，推薦使用VPC地址。
   AccessKey ID	阿里雲帳號或者RAM使用者的AccessKey ID和AccessKey Secret。
   AccessKey Secret

6. 測試資源群組連通性。

   建立資料來源時，需要測試資源群組的連通性，確保同步任務使用的資源群組能夠與資料來源正常連通，否則將無法正常執行資料同步任務。

   1. 在串連配置地區，單擊相應資源群組連通狀態列的測試連通性。

   2. 測試連通性通過後，連通狀態顯示可連通，單擊完成。可在資料來源列表中查看建立的資料來源。

      如果測試連通性結果為無法通過，可使用連通性診斷工具自助解決。

步驟二：新增OSS資料來源

配置OSS資料來源作為資料匯出的目標儲存。

1. 再次單擊新增資料來源，在對話方塊中搜尋並選擇資料來源類型為OSS，並配置相關的資料來源參數。

   參數	說明
   資料來源名稱	資料來源名稱必須以字母、數字、底線（_）組合，且不能以數字和底線（_）開頭。
   資料來源描述	對資料來源進行簡單描述，不得超過80個字元。
   訪問模式	RAM角色授權模式：DataWorks服務帳號以角色扮演的方式訪問資料來源。首次選擇該模式時，請根據頁面提示開啟授權。 Access Key模式：通過阿里雲帳號或者RAM使用者的AccessKey ID和AccessKey Secret訪問資料來源。
   選擇角色	僅當訪問模式為RAM角色授權模式時需要選擇RAM角色。
   AccessKey ID	僅當訪問模式為Access Key模式時需要填寫。阿里雲帳號或者RAM使用者的AccessKey ID和AccessKey Secret。
   AccessKey Secret
   地區	Bucket所在地區。
   Endpoint	OSS訪問網域名稱，詳見地區和Endpoint。
   Bucket	Bucket名稱。

2. 完成參數和連通性測試後，單擊完成添加資料來源。

步驟三：配置離線同步任務

建立並配置資料同步任務，定義從Table Store到OSS的資料轉送規則。

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                    "id",
                    "name",
                    "colName",
                    "version",
                    "colValue",
                    "opType",
                    "sequenceInfo"
                ],
                "startTimeString": "${startTime}",
                "table": "source_table",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "column": [
                    "0",
                    "1",
                    "2",
                    "3",
                    "4",
                    "5",
                    "6"
                ],
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_table.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

{
    "type": "job",
    "version": "2.0",
    "steps": [
        {
            "stepType": "otsstream",
            "parameter": {
                "statusTable": "TableStoreStreamReaderStatusTable",
                "maxRetries": 31,
                "isExportSequenceInfo": false,
                "datasource": "source_data",
                "column": [
                  {
                    "name": "_m_name"
                  },
                  {
                    "name": "_data_source"
                  },
                  {
                    "name": "_tags"
                  },
                  {
                    "name": "_time"
                  },
                  {
                    "name": "value",
                    "type": "int"
                  }
                ],
                "startTimeString": "${startTime}",
                "table": "source_series",
                "isTimeseriesTable":"true",
                "mode": "single_version_and_update_only",
                "endTimeString": "${endTime}"
            },
            "name": "Reader",
            "category": "reader"
        },
        {
            "stepType": "oss",
            "parameter": {
                "dateFormat": "yyyy-MM-dd HH:mm:ss",
                "datasource": "target_data",
                "writeSingleObject": false,
                "writeMode": "truncate",
                "encoding": "UTF-8",
                "fieldDelimiter": ",",
                "fileFormat": "csv",
                "object": "tablestore/source_series.csv"
            },
            "name": "Writer",
            "category": "writer"
        }
    ],
    "setting": {
        "errorLimit": {
            "record": "0"
        },
        "speed": {
            "concurrent": 2,
            "throttle": false
        }
    },
    "order": {
        "hops": [
            {
                "from": "Reader",
                "to": "Writer"
            }
        ]
    }
}

步驟四：查看同步結果

運行同步任務後，可通過日誌查看任務的執行狀態，並在OSS Bucket查看同步結果檔案。

1. 在頁面下方查看任務運行狀態和結果，出現以下資訊時表示同步任務運行成功。

   2025-11-18 11:16:23 INFO Shell run successfully!
   2025-11-18 11:16:23 INFO Current task status: FINISH
   2025-11-18 11:16:23 INFO Cost time is: 77.208s

2. 查看目標Bucket的檔案。

   前往Bucket列表，單擊目標Bucket，查看或下載同步結果檔案。