DataWorks:OTSStream data source

Prepare an OTSStream environment before data synchronization

SyncClient client = new SyncClient("", "", "", "");
# Method 1: Enable the Stream feature when you create a table.
CreateTableRequest createTableRequest = new CreateTableRequest(tableMeta);
createTableRequest.setStreamSpecification(new StreamSpecification(true, 24)); // The value 24 indicates that Tablestore retains incremental data for 24 hours. 
client.createTable(createTableRequest);
#Method 2: If you do not enable the Stream feature when you create a table, you can call the UpdateTable operation to enable this feature after the table is created.
UpdateTableRequest updateTableRequest = new UpdateTableRequest("tableName");
updateTableRequest.setStreamSpecification(new StreamSpecification(true, 24));
client.updateTable(updateTableRequest);

Supported read modes and data types

OTSStream Reader allows you to read incremental data from Tablestore in column or row mode. This section describes the data read processes and supported data types of the read modes.

Read data in column mode

In the multi-version mode of Tablestore, table data is organized in a three-level structure: row, column, and version. One row can have multiple columns, and the column name is not fixed. Each column can have multiple versions, and each version has a specific timestamp, which is the version number.

You can call the API operations of Tablestore to read and write data. Tablestore records the recent write and modify operations that are performed on table data to record the incremental data. Therefore, incremental data can be considered as a set of operation records.

Tablestore supports dynamic columns and the multi-version mode. Therefore, a row that is generated by OTSStream Reader is a version of a column rather than a row in Tablestore. After OTSStream Reader reads data from a row in Tablestore, OTSStream Reader converts the data to multiple rows. Each row includes the primary key value, column name, timestamp of the version for the column (version number), value of the version, and operation type. If the isExportSequenceInfo parameter is set to true, time series information is also included.

Read data in row mode

• Wide tables

  You can also use OTSStream Reader to read data in row mode. In this mode, OTSStream Reader reads operation records as rows. You must configure the mode parameter and specify the columns whose data you want to read.

  "parameter": {
    # Set mode to single_version_and_update_only, set isExportSequenceInfo to false, and configure other parameters, such as datasource and table, based on your business requirements. 
    "mode": "single_version_and_update_only", # The read mode. 
    "column":[  # The columns whose data you want to read from Tablestore. You can specify the columns based on your business requirements. 
            {
               "name": "uid"  # The name of a column, which can be a primary key column or a property column. 
            },
            {
               "name": "name"  # The name of a column, which can be a primary key column or a property column. 
            },
    ],
    "isExportSequenceInfo": false, # Specifies whether to read time series information. If you set the mode parameter to single_version_and_update_only, this parameter can only be set to false. 
  }

• Time series tables

  The Stream feature is automatically enabled when you create a time series table.

  OTSStream Reader allows you to read incremental data from a time series table. If the table is a time series table, you must configure the following parameters:

  "parameter": {
    # Configure the following parameters and other parameters, such as datasource and table, based on your business requirements. 
    "mode": "single_version_and_update_only", # The read mode. 
    "isTimeseriesTable":"true", # Specifies whether the table is a time series table. 
    "column":[  # The columns whose data you want to read from Tablestore. You can specify the columns based on your business requirements. 
            {
              "name": "_m_name" # The name of a metric name column. 
            },
            {
              "name": "_data_source" # The name of a data source column. 
            },
            {
              "name": "_tags" # The name of a tag column. Tags are converted into data of a string type. 
            },
            {
              "name": "tag1", # The name of a tag key. 
              "is_timeseries_tag":"true"  # Specifies whether the field is an internal field of the tags. 
            },
            {
              "name": "time" # The name of a timestamp column. 
            },
            {
               "name": "name" # The name of a property column. 
            },
    ],
    "isExportSequenceInfo": false, # Specifies whether to read time series information. If you set the mode parameter to single_version_and_update_only, this parameter can only be set to false. 
  }

  The data that is read by row is closer to the data in the original rows. This facilitates further data processing. If you read data by row, take note of the following points:

  • The rows that are read are extracted from operation records. Each row corresponds to a write or update operation. If you update only some of the columns for a row, the operation record contains only the columns that are updated.

  • The version number of each column, which is the timestamp of each column, cannot be read or deleted.

Data type mappings

Develop a data synchronization task

• For information about the configuration procedure, see Configure a batch synchronization task by using the codeless UI and Configure a batch synchronization task by using the code editor.

• For information about all parameters that are configured and the code that is run when you use the code editor to configure a batch synchronization task, see Appendix: Code and parameters.

Appendix: Code and parameters

Appendix: Configure a batch synchronization task by using the code editor

If you use the code editor to configure a batch synchronization task, you must configure parameters for the reader and writer of the related data source based on the format requirements in the code editor. For more information about the format requirements, see Configure a batch synchronization task by using the code editor. The following information describes the configuration details of parameters for the reader and writer in the code editor.

Code for OTSStream Reader

• Read data in column mode

  {
      "type":"job",
      "version":"2.0",// The version number. 
      "steps":[
          {
              "stepType":"otsstream",// The plug-in name. 
              "parameter":{
                  "datasource":"$srcDatasource",// The name of the data source. 
                  "dataTable":"",// The name of the table. 
                  "statusTable":"TableStoreStreamReaderStatusTable",// The name of the table that OTSStream Reader uses to store status records. 
                  "maxRetries":30,// The maximum number of retries for each request to read incremental data from Tablestore. Default value: 30. 
                  "isExportSequenceInfo":false,// Specifies whether to read time series information. 
                  "startTimeString":"${startTime}${hh}",// The start time of the incremental data, which is the same as the time when a synchronization task starts to run. Configure this parameter in the yyyymmddhh24miss format.
                  "endTimeString":"${endTime}${hh}"// The end time of the incremental data, which is the same as the time when the running of a synchronization task is complete. Configure this parameter in the yyyymmddhh24miss format.
              },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// The maximum number of dirty data records allowed. 
          },
          "speed":{
              "throttle":true,// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
              "concurrent":1,// The maximum number of parallel threads. 
              "mbps":"12"// The maximum transmission rate. Unit: MB/s. 
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• Read data from a wide table in row mode

  {
      "type":"job",
      "version":"2.0",// The version number. 
      "steps":[
          {
              "stepType":"otsstream",// The plug-in name. 
              "parameter":{
                  "datasource":"$srcDatasource",// The name of the data source. 
                  "dataTable":"",// The name of the table. 
                  "statusTable":"TableStoreStreamReaderStatusTable",// The name of the table that OTSStream Reader uses to store status records. 
                  "maxRetries":30,// The maximum number of retries for each request to read incremental data from Tablestore. Default value: 30. 
                  "isExportSequenceInfo":false,// Specifies whether to read time series information. 
                  "startTimeString":"${startTime}${hh}",// The start time of the incremental data, which is the same as the time when a synchronization task starts to run. Configure this parameter in the yyyymmddhh24miss format.
                  "endTimeString":"${endTime}${hh}"// The end time of the incremental data, which is the same as the time when the running of a synchronization task is complete. Configure this parameter in the yyyymmddhh24miss format.
                  "mode": "single_version_and_update_only",
                  "column":[
                          {
                              "name":"pId"
                          },
                          {
                              "name": "uId"
                          },
                          {
                              "name":"col0"
                          },
                          {
                              "name": "col1"
                          }
                      ],
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// The maximum number of dirty data records allowed. 
          },
          "speed":{
              "throttle":true,// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
              "concurrent":1,// The maximum number of parallel threads. 
              "mbps":"12" // The maximum transmission rate.
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

• Read data from a time series table in row mode

  {
      "type":"job",
      "version":"2.0",// The version number. 
      "steps":[
          {
              "stepType":"otsstream",// The plug-in name. 
              "parameter":{
                  "datasource":"$srcDatasource",// The name of the data source. 
                  "dataTable":"",// The name of the table. 
                  "statusTable":"TableStoreStreamReaderStatusTable",// The name of the table that OTSStream Reader uses to store status records. 
                  "maxRetries":30,// The maximum number of retries for each request to read incremental data from Tablestore. Default value: 30. 
                  "isExportSequenceInfo":false,// Specifies whether to read time series information. 
                  "startTimeString":"${startTime}${hh}",// The start time of the incremental data, which is the same as the time when a synchronization task starts to run. Configure this parameter in the yyyymmddhh24miss format.
                  "endTimeString":"${endTime}${hh}"// The end time of the incremental data, which is the same as the time when the running of a synchronization task is complete. Configure this parameter in the yyyymmddhh24miss format.
                  "mode": "single_version_and_update_only",
                  "isTimeseriesTable":"true",
                  "column": [
                            {
                              "name": "_m_name"
                            },
                            {
                              "name": "_data_source",
                            },
                            {
                              "name": "_tags",
                            },
                            {
                              "name": "string_column",
                            }
                      ]
                      },
              "name":"Reader",
              "category":"reader"
          },
          {
              "stepType":"stream",
              "parameter":{},
              "name":"Writer",
              "category":"writer"
          }
      ],
      "setting":{
          "errorLimit":{
              "record":"0"// The maximum number of dirty data records allowed. 
          },
          "speed":{
              "throttle":true,// Specifies whether to enable throttling. The value false indicates that throttling is disabled, and the value true indicates that throttling is enabled. The mbps parameter takes effect only when the throttle parameter is set to true. 
              "concurrent":1,// The maximum number of parallel threads. 
              "mbps":"12"// The maximum transmission rate. Unit: MB/s. 
  
          }
      },
      "order":{
          "hops":[
              {
                  "from":"Reader",
                  "to":"Writer"
              }
          ]
      }
  }

Parameters in code for OTSStream Reader

"column":[
    {"name":"pk1"},
	{"name":"col1"},
	{"name":"col2","dataType":"new"},
	{"name":"col2","dataType":"old"},
	{"name":"col2","dataType":"latest"}
],