同步增量数据到OSS

1. 在配置任务步骤的配置数据来源与去向区域，根据实际配置数据来源和数据去向。

   数据来源配置

   数据去向配置

   参数	说明
   表	表格存储中的数据表名称。
   开始时间	增量读取数据的开始时间和结束时间，分别配置为变量形式${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值的字符串标识符方式，可以是空字符串。
                   "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值的字符串标识符方式，可以是空字符串。
                   "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. 单击图标，保存配置。

   说明

   执行后续操作时，如果未保存脚本，则系统会出现保存确认的提示，单击确认即可。