Configure an OTSStream 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 Workspace Management > Data Source page.
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 parameters for the Tablestore connection.

   Parameter	Description
   Data Source Name	The name of the connection 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.
   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 ID	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 user interface (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, select the connection to the source data store.

   Parameter	Description
   Connection	The name of the connection.
   Table	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.
   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 an unpartitioned 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 upon each run. This rule is equivalent to the INSERT INTO statement.
   Convert Empty Strings to Null	Default value: No.

6. Configure field mappings.
   Each field in the source table on the left corresponds with a field in the destination table on the right. You can click Add to add a field, or move the pointer over a field and click the Delete icon to delete the field.
7. Configure channel control policies.
   
8. Click the Save icon in the toolbar.
9. Click the Run icon in the toolbar. You must configure custom parameters before you run the batch sync node.

Configure a batch sync node by using the code editor

To configure the batch sync node by using the code editor, click Switch to Code Editor in the toolbar, and then click OK.

{
  "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 isExportSequenceInfo 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.