Stream Kafka data to a time series table

Step 1: Deploy the Tablestore Sink Connector

1. You can obtain the Tablestore sink connector in one of the following ways.

   • Download and compile the source code from GitHub. The source code is available at Tablestore Sink Connector source code.

     1. Run the following command to download the Tablestore sink connector source code.

        git clone https://github.com/aliyun/kafka-connect-tablestore.git

     2. Go to the downloaded source code directory and run the following command to package the code using Maven.

        mvn clean package -DskipTests

        After the compilation is complete, the generated package, such as kafka-connect-tablestore-1.0.jar, is stored in the target directory.

   • Download the compiled kafka-connect-tablestore package.

2. Copy the package to the $KAFKA_HOME/libs directory on each node.

Step 2: Start the Tablestore Sink Connector

The Tablestore sink connector supports two running modes: standalone and distributed. You can choose a mode based on your requirements.

value.converter=org.apache.kafka.connect.json.JsonConverter
value.converter.schemas.enable=false

This section provides an example for standalone mode. The configuration for distributed mode is similar to syncing data to a standard data table. You must update the value converter settings in the connect-distributed.properties worker configuration file and adjust the time series settings in the connect-tablestore-sink-quickstart.json connector configuration file. For more information, see the steps for configuring distributed mode in Step 2: Start the Tablestore Sink Connector.

To configure standalone mode:

1. Modify the worker configuration file connect-standalone.properties and the connector configuration file connect-tablestore-sink-quickstart.properties based on your environment.

   • Example connect-standalone.properties worker configuration file

     The worker configuration includes parameters such as Kafka connection settings, serialization format, and offset commit frequency. This example is based on the official Kafka documentation. For more information, see Kafka Connect.

     # Licensed to the Apache Software Foundation (ASF) under one or more
     # contributor license agreements.  See the NOTICE file distributed with
     # this work for additional information regarding copyright ownership.
     # The ASF licenses this file to You under the Apache License, Version 2.0
     # (the "License"); you may not use this file except in compliance with
     # the License.  You may obtain a copy of the License at
     #
     #    http://www.apache.org/licenses/LICENSE-2.0
     #
     # Unless required by applicable law or agreed to in writing, software
     # distributed under the License is distributed on an "AS IS" BASIS,
     # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
     # See the License for the specific language governing permissions and
     # limitations under the License.
     # These are defaults. This file just demonstrates how to override some settings.
     bootstrap.servers=localhost:9092
     # The converters specify the format of data in Kafka and how to translate it into Connect data. Every Connect user will
     # need to configure these based on the format they want their data in when loaded from or stored into Kafka
     key.converter=org.apache.kafka.connect.json.JsonConverter
     value.converter=org.apache.kafka.connect.json.JsonConverter
     # Converter-specific settings can be passed in by prefixing the Converter's setting with the converter we want to apply
     # it to
     key.converter.schemas.enable=true
     value.converter.schemas.enable=false
     offset.storage.file.filename=/tmp/connect.offsets
     # Flush much faster than normal, which is useful for testing/debugging
     offset.flush.interval.ms=10000
     # Set to a list of filesystem paths separated by commas (,) to enable class loading isolation for plugins
     # (connectors, converters, transformations). The list should consist of top level directories that include 
     # any combination of: 
     # a) directories immediately containing jars with plugins and their dependencies
     # b) uber-jars with plugins and their dependencies
     # c) directories immediately containing the package directory structure of classes of plugins and their dependencies
     # Note: symlinks will be followed to discover dependencies or plugins.
     # Examples: 
     # plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors,
     #plugin.path=

   • Example connect-tablestore-sink-quickstart.properties connector configuration file

     The connector configuration includes parameters such as the connector class, Tablestore connection, and data mapping. For more information, see Configuration.

     # A unique name for the connector.
     name=tablestore-sink
     # The connector class.
     connector.class=TableStoreSinkConnector
     # The maximum number of tasks.
     tasks.max=1
     # A comma-separated list of topics to sink data from.
     topics=test
     # The following are Tablestore connection parameters.
     # The endpoint of the Tablestore instance.
     tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com
     # The authentication mode.
     tablestore.auth.mode=aksk
     # Your AccessKey ID and AccessKey secret. Required for 'aksk' authentication.
     tablestore.access.key.id=xxx
     tablestore.access.key.secret=xxx
     # The name of your Tablestore instance.
     tablestore.instance.name=xxx
     ## STS authentication settings. If you use STS, the following parameters are required. 
     ## You must also configure ACCESS_ID and ACCESS_KEY as environment variables.
     #sts.endpoint=
     #region=
     #account.id=
     #role.name=
     # A format string for the destination table name, which can include the  placeholder.
     # The topics.assign.tables setting takes precedence over this setting.
     # For example, if table.name.format is set to kafka_ and the Kafka topic name is 'test', the data is mapped to a Tablestore table named 'kafka_test'.
     table.name.format=<topic>
     # A mapping of topics to tables in the format ":". Use a colon (:) to separate the topic and table name, and a comma (,) to separate multiple mappings.
     # If this is not specified, the table.name.format setting is used.
     # topics.assign.tables=test:test_kafka
     # Specifies whether to automatically create the destination table. Default: false.
     auto.create=true
     # The following are settings for handling dirty data.
     # Errors can occur when parsing Kafka records or writing to a time series table. Use these settings to manage errors.
     # Specifies the fault tolerance level. Valid values: none, all. Default: none.
     # none: The connector task fails immediately on any error.
     # all: Skips records that cause errors and logs them.
     runtime.error.tolerance=none
     # The mode for handling dirty data. Valid values: ignore, kafka, tablestore. Default: ignore.
     # ignore: Ignores all errors.
     # kafka: Stores the problematic records and error information in a separate Kafka topic.
     # tablestore: Stores the problematic records and error information in a separate Tablestore data table.
     runtime.error.mode=ignore
     # If runtime.error.mode is set to 'kafka', configure the Kafka cluster address and topic name.
     # runtime.error.bootstrap.servers=localhost:9092
     # runtime.error.topic.name=errors
     # If runtime.error.mode is set to 'tablestore', configure the name of the Tablestore data table.
     # runtime.error.table.name=errors
     ## The following are settings specific to time series tables.
     # The connector's mode. Default: normal.
     tablestore.mode=timeseries
     # Mappings for the time series primary key fields.
     tablestore.timeseries.test.measurement=m
     tablestore.timeseries.test.dataSource=d
     tablestore.timeseries.test.tags=region,level
     # Mapping for the time series time field.
     tablestore.timeseries.test.time=timestamp
     tablestore.timeseries.test.time.unit=MILLISECONDS
     # Specifies whether to convert field column names to lowercase. Default: true. The TimeSeries model does not support uppercase letters in column names. If this is set to false and a column name contains uppercase letters, a write error occurs.
     tablestore.timeseries.toLowerCase=true
     # Specifies whether to store all non-primary key and non-timestamp fields as fields in the time series table. Default: true. If false, only the fields specified in tablestore.timeseries.test.field.name are stored.
     tablestore.timeseries.mapAll=true
     # The names of the field columns, separated by commas (,).
     tablestore.timeseries.test.field.name=cpu
     # The data types of the fields. Valid values: double, integer, string, binary, and boolean.
     # If multiple fields are specified, the types must correspond one-to-one with the field names. Separate multiple types with commas (,).
     tablestore.timeseries.test.field.type=double

2. Go to the $KAFKA_HOME directory and run the following command to start the connector in standalone mode.

   bin/connect-standalone.sh config/connect-standalone.properties config/connect-tablestore-sink-quickstart.properties

Step 3: Produce new records

1. Go to the $KAFKA_HOME directory and run the following command to start a console producer.

   bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test

   For a description of the parameters, see the following table.

   Configuration Item	Example	Description
   --broker-list	localhost:9092	The address and port of the Kafka cluster broker.
   --topic	test	The topic name. The topic is automatically created by default when the Tablestore Sink Connector starts. You can also create the topic manually.

2. Write new messages to the test topic.

   Important

   To import data into a time series table, you must write data to the topic in JSON format.

   {"m":"cpu","d":"127.0.0.1","region":"shanghai","level":1,"timestamp":1638868699090,"io":5.5,"cpu":"3.5"}

3. Log on to the Tablestore console to view the data.

   On the Time Series Table Management page, select the Data Management tab. Find the time series with the measurement name cpu, the data source 127.0.0.1, and the tags level=1 and region=shanghai. Click Query Data. The query returns a record where the time is 2021-12-07 17:18:19, cpu is 3.5, and io is 5.5.