DataWorks:Use Tablestore Stream Reader to configure a synchronization task

Add a data source

1. 1. Go to the Data Integration page.

      Log on to the DataWorks console. In the top navigation bar, select the desired region. In the left-side navigation pane, choose Data Integration > Data Integration. On the page that appears, select the desired workspace from the drop-down list and click Go to Data Integration.

2. In the left-side navigation pane of the Data Integration page, click Data Source. The Data Source page appears.

3. Click Add Data Source.

4. In the Add Data Source dialog box, click Tablestore.

5. In the Add OTS Data Source dialog box, configure the parameters.

   Parameter	Description
   Data Source Name	The name of the data source. The name can contain only letters, digits, and underscores (_) and must start with a letter.
   Data Source Description	The description of the data source. The description cannot exceed 80 characters in length.
   Endpoint	The endpoint of Tablestore.
   Tablestore Instance Name	The name of the Tablestore instance.
   AccessKey ID	The AccessKey ID of the Alibaba Cloud account that you use to connect to the Tablestore instance. You can copy the AccessKey ID on the AccessKey Pair page.
   AccessKey Secret	The AccessKey secret of the Alibaba Cloud account that you use to connect to the Tablestore instance.

6. Find the resource group that you want to use and click Test Network Connectivity in the Connection Status (Production Environment) column.

7. If the connectivity test is successful, click Complete.

Configure the batch synchronization task by using the codeless UI

1. Go to the DataStudio page.

   Log on to the DataWorks console. In the top navigation bar, select the desired region. In the left-side navigation pane, choose Data Development and O&M > Data Development. On the page that appears, select the desired workspace from the drop-down list and click Go to Data Development.

2. On the DataStudio page, find the desired workflow and click the name of the workflow. Right-click Data Integration and choose Create Node > Offline synchronization.

3. In the Create Node dialog box, configure the Name and Path parameters and click Confirm.

4. Select Tablestore Stream for Source and MaxCompute(ODPS) for Destination, select a resource group that you want to use to run the synchronization task, and then test the connectivity.

5. Configure the source and destination.

   Category	Parameter	Description
   Source	Table	The name of the table from which you want to read incremental data. You must make sure that the Stream feature is enabled for the table. You can enable the Stream feature for a table when you create the table.
   	Start time	The start time of the incremental data, in milliseconds. The start time is the left boundary of the left-closed, right-open time range of the incremental data. Configure this parameter in the yyyymmddhh24miss format.
   	End time	The end time of the incremental data, in milliseconds. The end time is the right boundary of the left-closed, right-open time range of the incremental data. Configure this parameter in the yyyymmddhh24miss format.
   	Status table	The name of the table that Tablestore Stream Reader uses to store status records.
   	Maximum number of retries	The maximum number of retries for each request to read incremental data from Tablestore. Default value: 30.
   	Export timing information	Specifies whether to read time series information. The time series information includes the time when data is written.
   Destination	Table	The table to which you want to write data.
   	Partition information	The table to which data will be written is a non-partitioned table. No partition key column is displayed.
   	Write Mode	Clear up existing data before writing (Insert Overwrite): All data in the table or partition is deleted before data import. This rule is equivalent to the INSERT OVERWRITE statement. Keep existing data before writing (Insert Into): No data is deleted before data import. New data is always appended. This rule is equivalent to the INSERT INTO statement.
   	Write by Converting Empty Strings into Null	Specifies whether to convert empty strings to null. Default value: No.

6. Configure field mappings.

   Fields in the source table on the left have a one-to-one mapping with fields in the destination table on the right. You can click Add a row to add a field. To remove a field, move the pointer over the field and click the Remove icon.

7. Configure channel control policies.

   

8. Click the Save icon in the top toolbar.

9. Click the Run icon in the top toolbar. You must configure custom parameters before you run the synchronization task.

Configure a synchronization task by using the code editor

{
  "type": "job",
  "version": "1.0",
  "configuration": {
    "reader": {
      "plugin": "Tablestore",
      "parameter": {
        "datasource": "Tablestore",// The name of the data source. Use the name of the data source that you have added. 
        "dataTable": "person",// The name of the table from which the incremental data is exported. You must make sure that the Stream feature is enabled for the table. You can enable the Stream feature for a table when you create the table. 
        "startTimeString": "${startTime}",// The start time (included) in milliseconds of the incremental data. Configure this parameter in the yyyymmddhh24miss format. 
        "endTimeString": "${endTime}",// The end time (excluded) in milliseconds of the incremental data. 
        "statusTable": "TableStoreStreamReaderStatusTable",// The name of the table that is used to store status records. 
        "maxRetries": 30,// The maximum number of retries for each request. 
        "isExportSequenceInfo": false,
      }
    },
    "writer": {
      "plugin": "odps",
      "parameter": {
        "datasource":"odps_first", // The name of the data source. 
        "table": "person",// The name of the destination table. 
        "truncate": true,
        "partition": "pt=${bdp.system.bizdate}",// The partition information. 
        "column": [// The names of the columns to which you want to write data. 
          "id",
          "colname",
          "version",
          "colvalue",
          "optype",
          "sequenceinfo"
        ]
      }
    },
    "setting": {
      "speed": {
        "mbps": 7,// The maximum transmission rate. Unit: MB/s. 
        "concurrent": 7// The maximum number of parallel threads. 
      }
    }
  }
}

If the isExportSequenceInfo parameter is set to true ("isExportSequenceInfo": true), the system exports an extra column for time series information. The time series information contains the time when data is written. The default value is false, which indicates that no time series information is exported.