Use OTSStream Reader to configure a sync node

Create a connection

1. Log on to the DataWorks console. Find the required workspace and click Data Integration.
   If you are using another service of DataWorks, click the More icon in the upper-left corner and choose All Products > Data Integration to go to the Data Integration page.
2. Click Connection in the left-side navigation pane to go to the Data Source page under Workspace Management.
3. Click New data source in the upper-right corner.
4. In the Add data source dialog box, select OTS as the connection type.
5. Set the parameters for the Tablestore connection.

   Parameter	Description
   Data Source Name	The name of the connection. The name can contain letters, digits, and underscores (_), and must start with a letter.
   Data source description	The description of the connection. The description can be up to 80 characters in length.
   Applicable environment	Valid values: Development and Production. Note This parameter is displayed only when the workspace is in standard mode.
   Endpoint	The endpoint of Tablestore.
   Table Store Instance name	The ID of the Tablestore instance.
   AccessKey ID	The AccessKey ID of the AccessKey pair. You can copy the AccessKey ID on the User Management page.
   AccessKey Secret	The AccessKey secret of the AccessKey pair, which is equivalent to the logon password.

6. Click Test connectivity.
7. After the connection passes the connectivity test, click Complete.

Configure a sync node on the codeless UI

1. On the Data Source page, click the More icon in the upper-left corner and choose All Products > Data Integration.
2. Click New offline synchronization on the homepage.
3. In the Create Node dialog box, set the Node Name and Location parameters and click Commit.
4. On the configuration tab of the batch sync node, configure the connection to the source data store.

   Parameter	Description
   Connection	The name of the connection.
   Table	The name of the table from which incremental data is exported. You must enable the Stream feature for the table when you create the table, or by calling the UpdateTable operation after you create the table.
   Start Timestamp	The start time (included) in milliseconds of the incremental data. The format is yyyymmddhh24miss.
   End Timestamp	The end time (excluded) in milliseconds of the incremental data. The format is yyyymmddhh24miss.
   State Table	The name of the table that is used to store status records.
   Maximum Retries	The maximum number of retries of each request that is used to read incremental data from Tablestore. Default value: 30.
   Export Time Information	Specifies whether to export time series information, including the time when data is written.

5. Configure the connection to the destination data store.

   Parameter	Description
   Connection	The name of the connection that you have configured.
   Table	The table that you want to synchronize.
   Partition Key Column	The table to be synchronized is a non-partitioned table. No partition key column is displayed.
   Writing Rule	Write with Original Data Deleted (Insert Overwrite): All data in the table or partition is deleted before data import. This rule is equivalent to the INSERT OVERWRITE statement. Write with Original Data Retained (Insert Into): No data is deleted before data import. New data is always appended. This rule is equivalent to the INSERT INTO statement.
   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 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 set custom parameters before you run the sync node.

Configure a sync node by using the code editor

To configure the sync node by using the code editor, click the Switch to Code Editor icon in the top toolbar and click OK. The code editor appears.

{
  "type": "job",
  "version": "1.0",
  "configuration": {
    "reader": {
      "plugin": "otsstream",
      "parameter": {
        "datasource": "otsstream", // The name of the connection. Use the name of the connection that you have created.
        "dataTable": "person", // The name of the table from which the incremental data is exported. You must enable the Stream feature for the table when you create the table, or by calling the UpdateTable operation after you create the table.
        "startTimeString": "${startTime}", // The start time (included) in milliseconds of the incremental data. The format is yyyymmddhh24miss.
        "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 of each request.
        "isExportSequenceInfo": false,
      }
    },
    "writer": {
      "plugin": "odps",
      "parameter": {
        "datasource":"odps_first", // The name of the connection.
        "table": "person", // The name of the destination table.
        "truncate": true,
        "partition": "pt=${bdp.system.bizdate}", // The information about the partition.
        "column": [ // The column to which data is written.
          "id",
          "colname",
          "version",
          "colvalue",
          "optype",
          "sequenceinfo"
        ]
      }
    },
    "setting": {
      "speed": {
        "mbps": 7, // The maximum transmission rate.
        "concurrent": 7 // The maximum number of concurrent 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.