Tablestore:Synchronize incremental data to OSS

(Recommended) Use the codeless UI

1. In the Configure Source and Destination section of the Configure tasks step, configure the source and destination data sources based on your business requirements.

   Configure the source data source

   Parameter	Description
   Table	The name of the data table in Tablestore.
   Start Time	The start time and end time of the time range during which incremental data is read. Specify the parameters in the ${startTime} and ${endTime} formats. The specific formats are configured in scheduling properties. The time range is a left-closed, right-open interval.
   End Time
   State Table	The name of the table that stores the status of Tablestore Stream. Default value: TableStoreStreamReaderStatusTable.
   Maximum Retries	The maximum number of retries allowed for each request to read incremental data from Tablestore.
   Export Time Series Information	Specifies whether to export time series information. The time series information includes the time when data is written.

   Configure the destination data source

   Parameter	Description
   Text type	The format of the object that is written to OSS, such as .csv or .txt. Note Different object formats require different configurations. Select an object format and specify the specific parameters that are displayed.
   File name (including path)	This parameter is displayed only when you set the Text type parameter to text, csv, or orc. The name of the object in OSS. The name can contain a path, such as tablestore/20231130/myotsdata.csv.
   path	This parameter is displayed only when you set the Text type parameter to parquet. The path of the object in OSS, such as tablestore/20231130/.
   fileName	This parameter is displayed only when you set the Text type parameter to parquet. The name of the object in OSS.
   Column Delimiter	This parameter is displayed only when you set the Text type parameter to text or csv. The delimiter that is used to separate columns when data is written to the OSS object.
   Row Delimiter	This parameter is displayed only when you set the Text type parameter to text. The custom delimiter that is used to separate rows. Example: \u0001. You must use a delimiter that does not exist in the data as the row delimiter. If you want to use the default row delimiter \n for Linux or \r\n for Windows, we recommend that you leave this parameter empty. The system automatically reads data based on the default row delimiter.
   Coding	This parameter is displayed only when you set the Text type parameter to text or csv. The encoding format of the OSS object to which you write data.
   null value	This parameter is displayed only when you set the Text type parameter to text, csv, or orc. The string that can be interpreted as null in the source data source. For example, if you set this parameter to null and the value of a field in the source data source is null, the system recognizes the field as null.
   Time format	This parameter is displayed only when you set the Text type parameter to text or csv. The time format in which date data is written to the OSS object, such as yyyy-MM-dd.
   Prefix conflict	The processing method that is used when the specified object name is the same as the name of an existing object in OSS. Valid values: Replace: deletes the existing object and creates another object of the same name. Retain: retains the existing object and creates another object whose name combines the existing object name and a random suffix. Error: reports an error and stops the batch synchronization task.
   Split File	This parameter is displayed only when you set the Text type parameter to text or csv. The maximum size of a single OSS object to which data is written. Unit: MB. The maximum size is 100 GB. When the size of an object that is written reaches the value of this parameter, the system generates a new object to continue writing data until all data is written.
   Write as Single File	This parameter is displayed only when you set the Text type parameter to text or csv. Specifies whether to write a single object to OSS at a time. By default, multiple objects are written to OSS at a time. If no data is read during data writing, the following cases may occur: If an object header is configured, an empty object that contains only the object header is generated. If no object header is configured, a completely empty object is generated. If you want to write a single object to OSS at a time, select Write as Single File. In this case, if no data is read during data writing, no empty file is generated.
   First Row as Table Header	This parameter is displayed only when you set the Text type parameter to text or csv. Specifies whether to write the first row as table headers when data is written to an object. By default, no table headers are generated. If you want to write the first row as table headers, select First Row as Table Header.

2. In the Field Mappings section, the system automatically completes field mappings. In this section, retain the default settings.

   The source fields include the primary key columns and incremental data changes in the data table. The destination fields cannot be configured.

   

3. In the Channel Control section, configure the parameters for task execution, such as the Task Expected Maximum Concurrency, Synchronization rate, Policy for Dirty Data Records, and Distributed Execution parameters. For more information about the parameters, see Configure channel control policies.

4. Click the icon to save the configurations.

   Note

   If you do not save the configurations, a message prompting you to save the configurations appears when you perform subsequent operations. Click OK to save the configurations.

Use the code editor

To synchronize incremental data, you must use OTSStream Reader and OSS Writer. For more information about how to configure the task by using the code editor, see Tablestore Stream data source and OSS data source.

1. In the Configure tasks step, click the icon. In the message that appears, click OK.

2. In the code editor, specify the parameters based on the following sample code.

   Important

   Comments are provided in the sample code to help you understand the configurations. Delete all comments when you use the sample code.

   Export incremental data based on the row mode

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", // The name of the reader. You cannot change the value of this parameter. 
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", // The name of the table that stores the status of Tablestore Stream. In most cases, you do not need to change the value of this parameter. 
                   "maxRetries": 30, // The maximum number of retries allowed for each request. 
                   "isExportSequenceInfo": false, // Specifies whether to export time series information. The time series information includes the time when data is written. 
   				        "mode": "single_version_and_update_only", // The mode in which Tablestore Stream is used to export data. Set this parameter to single_version_and_update_only. If the configuration template does not contain this parameter, add this parameter. 
                   "datasource": "otssource", // The name of the source data source. Specify this parameter based on your business requirements. 
                   "envType": 1,
                   "column": [ 
   				          {
                        "name": "pk1" 
                     },
                     {
                        "name": "pk2"  
                     },
                     {
                        "name": "col1"  
                     }
                   ],
                   "startTimeString": "${startTime}", // The start time of data export. The task must be started in loops because this task is used for incremental data export. The start time for each loop varies. Therefore, you must use a variable, such as ${startTime}. 
                   "table": "mytable", // The name of the source table in Tablestore. 
                   "endTimeString": "${endTime}" // The end time of data export. You must use a variable, such as ${endTime}. 
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", // The name of the writer. You cannot change the value of this parameter. 
               "parameter": { // If you specify parameters based on this sample code, data can be exported only in the CSV and TEXT formats. If you want to export data in the Parquet or ORC format, specify parameters based on the content in the "OSS data source" topic. 
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", // The string used to identify the null field value. The value can be an empty string. 
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", // The format of the time. 
                   "datasource": "osssource", // The name of the OSS data source. Specify this parameter based on your business requirements. 
                   "envType": 1,
                   "writeSingleObject": true, // Specifies whether to write a single file to OSS at a time. A value of true specifies that you write a single file to OSS at a time. If no data is read, no empty file is generated. A value of false specifies that you write multiple files to OSS at a time. If no data is read and a file header is configured, an empty file that contains only the file header is generated. Otherwise, an empty file is generated. 
                   "writeMode": "truncate", // The operation to be performed by the system if an object with the specified source file name exists in the destination data source. Valid values: truncate, append, and nonConflict. A value of truncate specifies that the system clears the object in the destination data source. A value of append specifies that the system appends the data to the object in the destination data source. A value of nonConflict specifies that an error is reported.  
                   "encoding": "UTF-8", // The encoding type. 
                   "fieldDelimiter": ",", // The delimiter used to separate columns. 
                   "fileFormat": "csv", // The file format. Valid values: csv, text, parquet, and orc. 
                   "object": "" // The prefix of the name of the file that you want to synchronize to OSS. We recommend that you use the "Tablestore instance name/Table name/Date" format. Example: "instance/table/{date}". 
               },
               "name": "Writer",
               "category": "writer" 
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" // The number of errors allowed. If the number of errors exceeds this value, the synchronization task fails. 
           },
           "locale": "zh",
           "speed": {
               "throttle": false, // Specifies whether to enable throttling. A value of false specifies that throttling is disabled, and a value of true specifies that throttling is enabled. The mbps parameter takes effect only if the throttle parameter is set to true. 
               "concurrent": 3 // The maximum number of concurrent tasks. 
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

   Export incremental data based on the incremental changes in the column values of rows

   {
       "type": "job",
       "version": "2.0",
       "steps": [
           {
               "stepType": "otsstream", // The name of the reader. You cannot change the value of this parameter. 
               "parameter": {
                   "statusTable": "TableStoreStreamReaderStatusTable", // The name of the table that stores the status of Tablestore Stream. In most cases, you do not need to change the value of this parameter. 
                   "maxRetries": 30, // The maximum number of retries allowed for each request. 
                   "isExportSequenceInfo": false, // Specifies whether to export time series information. The time series information includes the time when data is written. 
                   "datasource": "otssource",
                   "envType": 1,
                   "column": [ // The columns that you want to export from the data table to OSS. If the configuration template does not contain this parameter, add this parameter. In this example, the default settings are used. 
                       "pk1", // The name of the primary key column. If the data table has multiple primary key columns, you must specify all of them. 
                       "pk2", // The name of the primary key column. If the data table has multiple primary key columns, you must specify all of them. 
                       "colName", // The name of the attribute column whose value is incremental data. You do not need to change the value of this parameter. 
                       "version", // The data version number of the column whose value is incremental data. You do not need to change the value of this parameter. The value of this parameter is a 64-bit timestamp. Unit: millisecond. 
                       "colValue", // The value of the attribute column whose value is incremental data. You do not need to change the value of this parameter. 
                       "opType", // The type of the incremental data operation. You do not need to change the value of this parameter. 
                       "sequenceInfo" // The ID of the auto-increment sequence. You do not need to change the value of this parameter. 
                   ],
                   "startTimeString": "${startTime}", // The start time of data export. The task must be started in loops because this task is used for incremental data export. The start time for each loop varies. Therefore, you must use a variable, such as ${startTime}. 
                   "table": "mytable", // The name of the source table in Tablestore. 
                   "endTimeString": "${endTime}" // The end time of data export. You must use a variable, such as ${endTime}. 
               },
               "name": "Reader",
               "category": "reader"
           },
           {
               "stepType": "oss", // The name of the writer. You cannot change the value of this parameter. 
               "parameter": { // If you specify parameters based on this sample code, data can be exported only in the CSV and TEXT formats. If you want to export data in the Parquet or ORC format, specify parameters based on the content in the "OSS data source" topic. 
                   "fieldDelimiterOrigin": ",",
                   "nullFormat": "null", // The string used to identify the null field value. The value can be an empty string. 
                   "dateFormat": "yyyy-MM-dd HH:mm:ss", // The format of the time. 
                   "datasource": "osssource", // The name of the OSS data source. Specify this parameter based on your business requirements. 
                   "envType": 1,
                   "writeSingleObject": true, // Specifies whether to write a single file to OSS at a time. A value of true specifies that you write a single file to OSS at a time. If no data is read, no empty file is generated. A value of false specifies that you write multiple files to OSS at a time. If no data is read and a file header is configured, an empty file that contains only the file header is generated. Otherwise, an empty file is generated. 
                   "column": [ // The columns that you want to synchronize to OSS. Specify this parameter by using the sequence number of a column. You do not need to change the value of this parameter. 
                       "0",
                       "1",
                       "2",
                       "3",
                       "4",
                       "5",
                       "6"
                   ],
                   "writeMode": "truncate", // The operation to be performed by the system if an object with the specified source file name exists in the destination data source. Valid values: truncate, append, and nonConflict. A value of truncate specifies that the system clears the object in the destination data source. A value of append specifies that the system appends the data to the object in the destination data source. A value of nonConflict specifies that an error is reported.  
                   "encoding": "UTF-8", // The encoding type. 
                   "fieldDelimiter": ",", // The delimiter used to separate columns. 
                   "fileFormat": "csv", // The file format. Valid values: csv and text. 
                   "object": "" // The prefix of the name of the file that you want to synchronize to OSS. We recommend that you use the "Tablestore instance name/Table name/Date" format. Example: "instance/table/{date}". 
               },
               "name": "Writer",
               "category": "writer"
           },
           {
               "copies": 1,
               "parameter": {
                   "nodes": [],
                   "edges": [],
                   "groups": [],
                   "version": "2.0"
               },
               "name": "Processor",
               "category": "processor"
           }
       ],
       "setting": {
           "errorLimit": {
               "record": "0" // The number of errors allowed. If the number of errors exceeds this value, the synchronization task fails. 
           },
           "locale": "zh",
           "speed": {
               "throttle": false, // Specifies whether to enable throttling. A value of false specifies that throttling is disabled, and a value of true specifies that throttling is enabled. The mbps parameter takes effect only if the throttle parameter is set to true. 
               "concurrent": 3 // The maximum number of concurrent tasks. 
           }
       },
       "order": {
           "hops": [
               {
                   "from": "Reader",
                   "to": "Writer"
               }
           ]
       }
   }

3. Click the icon to save the configurations.

   Note

   If you do not save the script, a message prompting you to save the script appears when you perform subsequent operations. Click OK to save the script.