Tablestore:Configuration description

Configuration examples

The configuration items vary depending on whether data is synchronized from Kafka to a data table or a time series table in Tablestore. The configuration examples of the configuration files vary based on the working mode. This section provides an example on how to configure data synchronization from Kafka to a data table in Tablestore. To synchronize data to a time series table in Tablestore, you need to add configuration items that are specific to data synchronization from Kafka to a time series table in Tablestore.

• The following sample code provides an example on how to configure the configuration file in the .properties format for Tablestore Sink Connector in the standalone mode

  # Specify the connector name. 
  name=tablestore-sink
  # Specify the connector class. 
  connector.class=TableStoreSinkConnector
  # Specify the maximum number of tasks. 
  tasks.max=1
  # Specify the list of Kafka topics from which data is exported. 
  topics=test
  # Specify values for the following Tablestore connection parameters: 
  # The endpoint of the Tablestore instance.  
  tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
  # The AccessKey pair which consists of an AccessKey ID and an AccessKey secret. 
  tablestore.access.key.id=xxx
  tablestore.access.key.secret=xxx
  # The name of the Tablestore instance. 
  tablestore.instance.name=xxx
  
  # Specify the following data mapping parameters: 
  # Specify the parser that is used to parse Kafka message records. 
  # The DefaultEventParser of Tablestore Sink Connector supports the Struct and Map classes of Kafka Connect. You can also use a custom EventParser. 
  event.parse.class=com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser
  
  # Specify the format string for the name of the destination Tablestore table. <topic> can be used in the string as a placeholder for the topic from which you want to export data. 
  # topics.assign.tables is assigned a higher priority than table.name.format. If topics.assign.tables is specified, ignore the configuration of table.name.format. 
  # For example, if table.name.format is set to kafka_<topic> and the name of the Kafka topic from which you want to export data is test, Kafka message records from the test topic are mapped to the table named kafka_test in Tablestore. 
  table.name.format=<topic>
  # Specify the mapping between the Kafka topic and the destination Tablestore table. The value must be in the <topic>:<tablename> format. The topic name and table name are separated with a colon (:). If you want to specify multiple mappings, separate them with commas (,). 
  # If the mapping is not specified, the configuration of table.name.format is used. 
  # topics.assign.tables=test:test_kafka
  
  # Specify the primary key mode. Valid values: kafka, record_key, and record_value. Default value: kafka. 
  # kafka: <connect_topic>_<connect_partition> and <connect_offset> are used as the primary key of the data table. 
  # record_key: Fields in the record keys are used as the primary key of the data table. 
  # record_value: Fields in the record values are used as the primary key of the data table. 
  primarykey.mode=kafka
  
  # Specify the name and data type of the primary key column in the destination Tablestore data table. 
  # The format of the primary key column name is tablestore.<tablename>.primarykey.name. The format of the data type of the primary key column is tablestore.<tablename>.primarykey.type. 
  # <tablename> is a placeholder for the data table name. 
  # If the primary key mode is kafka, you do not need to specify the name and data type of the primary key column. The default primary key column names {"topic_partition","offset"} and the default data types {string, integer} of the primary key columns are used. 
  # If the primary key mode is record_key or record_value, you must specify the name and data type of the primary key column. 
  # tablestore.test.primarykey.name=A,B
  # tablestore.test.primarykey.type=string,integer
  
  # Specify an attribute column whitelist to filter the fields in the record values to obtain the required attribute columns. 
  # By default, the attribute column whitelist is empty. All fields in the record values are used as the attribute columns of the data table. 
  # The format of the attribute column name is tablestore.<tablename>.columns.whitelist.name. The format of the data type of the attribute column is tablestore.<tablename>.columns.whitelist.type. 
  # <tablename> is a placeholder for the data table name. 
  # tablestore.test.columns.whitelist.name=A,B
  # tablestore.test.columns.whitelist.type=string,integer
  
  # Specify how to write Kafka message records to the destination Tablestore table: 
  # Specify the write mode. Valid values: put and update. Default value: put. 
  # put: Data in the destination table is overwritten by Kafka message records. 
  # update: Data in the destination table is updated by Kafka message records. 
  insert.mode=put
  # Specify whether to write data in the sequence that data is read. Default value: true. You can disable this option to improve the write performance. 
  insert.order.enable=true
  # Specify whether to automatically create a destination table. Default value: false. 
  auto.create=false
  
  # Specify the delete mode. Valid values: none, row, column, and row_and_column. Default value: none. 
  # none: No delete operations can be performed. 
  # row: Rows can be deleted. 
  # column: Attribute columns can be deleted. 
  # row_and_column: Rows and attribute columns can be deleted. 
  delete.mode=none
  
  # Specify the maximum number of rows that can be included in the buffer queue in the memory when data is written to the data table. Default value: 1024. The value of this parameter must be an exponent of 2. 
  buffer.size=1024
  # Specify the number of callback threads that are used when data is written to the data table. Default value = Number of vCPUs + 1. 
  # max.thread.count=
  # Specify the maximum number of concurrent write requests that can be sent to write data to the data table. Default value: 10. 
  max.concurrency=10
  # Specify the number of buckets to which data is written. Default value: 3. If you increase the value of this parameter, the concurrent write capability can be increased. However, you cannot set the value of this parameter to a value greater than the maximum number of concurrent write requests that you specified. 
  bucket.count=3
  # Specify the interval at which the buffer queue is refreshed when data is written to the data table. Unit: milliseconds. Default value: 10000. 
  flush.Interval=10000
  
  # Specify how to process dirty data: 
  # An error may occur when the Kafka message records are parsed or written to the data table. You can specify the following two parameters to determine how to fix the error: 
  # Specify the fault tolerance capability. Valid values: none and all. Default value: none. 
  # none: An error causes the data import task that uses Tablestore Sink Connector to fail. 
  # all: The message records for which errors are reported are skipped and logged. 
  runtime.error.tolerance=none
  # Specify how dirty data is logged. Valid values: ignore, kafka, and tablestore. Default value: ignore. 
  # ignore: All errors are ignored. 
  # kafka: The message records for which errors are reported and the error messages are stored in a different Kafka topic. 
  # tablestore: The message records for which errors are reported and the error messages are stored in a different Tablestore data table. 
  runtime.error.mode=ignore
  
  # If you set runtime.error.mode to kafka, you must specify the Kafka cluster address and the topic. 
  # runtime.error.bootstrap.servers=localhost:9092
  # runtime.error.topic.name=errors
  
  # If you set runtime.error.mode to tablestore, you must specify the name of the Tablestore data table. 
  # runtime.error.table.name=errors

• The following sample code provides an example on how to configure the configuration file in the .json format for Tablestore Sink Connector in the distributed mode:

  {
    "name": "tablestore-sink",
    "config": {
      // Specify the connector class. 
      "connector.class":"TableStoreSinkConnector",
      // Specify the maximum number of tasks. 
      "tasks.max":"3",
      // Specify the list of Kafka topics from which you want to export data. 
      "topics":"test",
      // Specify values for the following Tablestore connection parameters: 
      // The endpoint of the Tablestore instance. 
      "tablestore.endpoint":"https://xxx.xxx.ots.aliyuncs.com",
      // The AccessKey pair which consists of an AccessKey ID and an AccessKey secret. 
      "tablestore.access.key.id":"xxx",
      "tablestore.access.key.secret":"xxx",
      // The name of the Tablestore instance. 
      "tablestore.instance.name":"xxx",
  
      // Specify the following data mapping parameters: 
      // Specify the parser that is used to parse Kafka message records. 
      // The DefaultEventParser of Tablestore Sink Connector supports the Struct and Map classes of Kafka Connect. You can also use a custom EventParser. 
      "event.parse.class":"com.aliyun.tablestore.kafka.connect.parsers.DefaultEventParser",
  
      // Specify the format string for the name of the destination Tablestore table. <topic> can be used in the string as a placeholder for the topic from which you want to export data. 
      // topics.assign.tables is assigned a higher priority than table.name.format. If topics.assign.tables is specified, ignore the configuration of table.name.format. 
      // For example, if table.name.format is set to kafka_<topic> and the name of the Kafka topic from which you want to export data is test, Kafka message records from the test topic are mapped to the table named kafka_test in Tablestore. 
      "table.name.format":"<topic>",
      // Specify the mapping between the Kafka topic and the destination Tablestore table. The value must be in the <topic>:<tablename> format. The topic name and table name are separated with a colon (:). If you want to specify multiple mappings, separate them with commas (,). 
      // If the mapping is not specified, the configuration of table.name.format is used. 
      // "topics.assign.tables":"test:test_kafka",
  
      // Specify the primary key mode. Valid values: kafka, record_key, and record_value. Default value: kafka. 
      // kafka: <connect_topic>_<connect_partition> and <connect_offset> are used as the primary key of the data table. 
      // record_key: Fields in the record keys are used as the primary key of the data table. 
      // record_value: Fields in the record values are used as the primary key of the data table. 
      "primarykey.mode":"kafka",
  
      // Specify the name and data type of the primary key column in the destination Tablestore data table. 
      // The format of the primary key column name is tablestore.<tablename>.primarykey.name. The format of the data type of the primary key column is tablestore.<tablename>.primarykey.type. 
      // <tablename> is a placeholder for the data table name. 
      // If the primary key mode is kafka, you do not need to specify the name and data type of the primary key column. The default primary key column names {"topic_partition","offset"} and the default data types {string, integer} of the primary key columns are used. 
      // If the primary key mode is record_key or record_value, you must specify the name and data type of the primary key column. 
      // "tablestore.test.primarykey.name":"A,B",
      // "tablestore.test.primarykey.type":"string,integer",
  
      // Specify an attribute column whitelist to filter the fields in the record values to obtain the required attribute columns. 
      // By default, the attribute column whitelist is empty. All fields in the record values are used as the attribute columns of the data table. 
      // The format of the attribute column name is tablestore.<tablename>.columns.whitelist.name. The format of the data type of the attribute column is tablestore.<tablename>.columns.whitelist.type. 
      // <tablename> is a placeholder for the data table name. 
      // "tablestore.test.columns.whitelist.name":"A,B",
      // "tablestore.test.columns.whitelist.type":"string,integer",
  
      // Specify how to write Kafka message records to the destination Tablestore table: 
      // Specify the write mode. Valid values: put and update. Default value: put. 
      // put: Data in the table is overwritten by Kafka message records. 
      // update: Data in the table is updated by Kafka message records. 
      "insert.mode":"put",
      // Specify whether to write data in the sequence that data is read. Default value: true. You can disable this option to improve the write performance. 
      "insert.order.enable":"true",
      // Specify whether to automatically create a destination table. Default value: false. 
      "auto.create":"false",
  
      // Specify the delete mode. Valid values: none, row, column, and row_and_column. Default value: none. 
      // none: No delete operations can be performed. 
      // row: Rows can be deleted. 
      // column: Attribute columns can be deleted. 
      // row_and_column: Rows and attribute columns can be deleted. 
      "delete.mode":"none",
  
      // Specify the maximum number of rows that can be included in the buffer queue in the memory when data is written to the data table. Default value: 1024. The value of this parameter must be an exponent of 2. 
      "buffer.size":"1024",
      // Specify the number of callback threads that are used when data is written to the data table. Default value = Number of vCPUs + 1. 
      // "max.thread.count":
      // Specify the maximum number of concurrent write requests that can be sent to write data to the data table. Default value: 10. 
      "max.concurrency":"10",
      // Specify the number of buckets to which data is written. Default value: 3. You can increase the value of this parameter to increase the concurrent write capability. However, you cannot set the value of this parameter to a value greater than the maximum number of concurrent write requests that you specified. 
      "bucket.count":"3",
      // Specify the interval at which the buffer queue is refreshed when data is written to the data table. Unit: milliseconds. Default value: 10000. 
      "flush.Interval":"10000",
  
      // Specify how to process dirty data: 
      // An error may occur when the Kafka message records are parsed or written to the data table. You can specify the following two parameters to determine how to fix the error: 
      // Specify the fault tolerance capability. Valid values: none and all. Default value: none. 
      // none: An error causes the data import task that uses Tablestore Sink Connector to fail. 
      // all: The message records for which errors are reported are skipped and logged. 
      "runtime.error.tolerance":"none",
      // Specify how dirty data is logged. Valid values: ignore, kafka, and tablestore. Default value: ignore. 
      // ignore: All errors are ignored. 
      // kafka: The message records for which errors are reported and the error messages are stored in a different Kafka topic. 
      // tablestore: The message records for which errors are reported and the error messages are stored in a different Tablestore data table. 
      "runtime.error.mode":"ignore"
  
      // If you set runtime.error.mode to kafka, you must specify the Kafka cluster address and the topic. 
      // "runtime.error.bootstrap.servers":"localhost:9092",
      // "runtime.error.topic.name":"errors",
  
      // If you set runtime.error.mode to tablestore, you must specify the name of the Tablestore data table. 
      // "runtime.error.table.name":"errors",
    }

Parameters

The following table describes the parameters in the configuration file. You need to configure time series-related parameters only when you synchronize data from Kafka to a time series table in Tablestore.

connector.class=com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector

Appendix: Data type mappings between Kafka and Tablestore

The following table describes the mappings between the data types of Kafka and Tablestore.

Appendix: Delete syntax

The following table describes the methods that are used to write data to a Tablestore data table based on the configurations of the write mode (insert.mode) and delete mode (delete.mode) when message records contain empty values and data is synchronized from Kafka to a data table in Tablestore.