同步增量数据到OSS

步骤一：新增表格存储数据源

在DataWorks中配置表格存储数据源，建立与源数据的连接。

1. 登录DataWorks控制台，切换至目标地域后，单击左侧导航栏的数据集成 > 数据集成，在下拉框中选择对应工作空间后单击进入数据集成。

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. 完成参数和连通性测试后，单击完成添加数据源。

步骤三：配置离线同步任务

创建并配置数据同步任务，定义从表格存储到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，查看或下载同步结果文件。