Configure Kafka for time series tables - Tablestore: Central de Documentação do Alibaba Cloud

Use o pacote kafka-connect-tablestore para gravar dados do Kafka em uma tabela de séries temporais do Tablestore. Este tópico descreve como configurar esse sink de dados.

Pré-requisitos

O Kafka deve estar instalado, com o ZooKeeper e o Kafka em execução. Para mais informações, consulte a documentação oficial do Kafka.
Ative o serviço Tablestore e crie uma instância e uma tabela de séries temporais. Para mais informações, consulte Introdução ao modelo TimeSeries.

Nota
Você também pode configurar o Tablestore Sink Connector para criar automaticamente a tabela de séries temporais de destino definindo o parâmetro auto.create como true.
Obtenha um AccessKey. Para mais informações, consulte Criar um AccessKey.

Informações básicas

O Tablestore armazena e analisa dados de séries temporais. Para mais informações, consulte Visão geral do modelo TimeSeries.

Etapa 1: Implantar o Tablestore Sink Connector

Obtenha o Tablestore Sink Connector de uma das seguintes formas:
- Baixe e compile o código source do GitHub. O código source está disponível em Código source do Tablestore Sink Connector.
  1. Execute o comando a seguir para baixar o código source do Tablestore Sink Connector.
    git clone https://github.com/aliyun/kafka-connect-tablestore.git
  2. Acesse o diretório do código source baixado e execute o comando a seguir para empacotar o código com o Maven.
    mvn clean package -DskipTests
    Após a conclusão da compilação, o pacote gerado, como kafka-connect-tablestore-1.0.jar, fica armazenado no diretório target.
- Baixe o pacote kafka-connect-tablestore já compilado.
Copie o pacote para o diretório $KAFKA_HOME/libs em cada nó.

Etapa 2: Iniciar o Tablestore Sink Connector

O Tablestore Sink Connector oferece dois modos de execução: standalone e distribuído. Escolha o modo adequado às suas necessidades.

Ao gravar dados de séries temporais, os registros de mensagens do Kafka devem estar no formato JSON. Portanto, use o JsonConverter ao iniciar o Tablestore Sink Connector. Não é necessário extrair um schema ou fornecer uma chave. Configure os parâmetros a seguir nos arquivos connect-standalone.properties e connect-distributed.properties conforme indicado abaixo.

Nota

Se você fornecer uma chave, configure os parâmetros key.converter e key.converter.schemas.enable de acordo com o formato da chave.

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

Esta seção apresenta um exemplo para o modo standalone. A configuração do modo distribuído assemelha-se à sincronização de dados para uma tabela de dados padrão. Atualize as definições do conversor de valores no arquivo de configuração do worker connect-distributed.properties e ajuste as configurações de séries temporais no arquivo de configuração do conector connect-tablestore-sink-quickstart.json. Para mais informações, consulte as etapas de configuração do modo distribuído em Etapa 2: Iniciar o Tablestore Sink Connector.

Para configurar o modo standalone:

Modifique o arquivo de configuração do worker connect-standalone.properties e o arquivo de configuração do conector connect-tablestore-sink-quickstart.properties conforme o seu ambiente.

Exemplo de arquivo de configuração do worker connect-standalone.properties

A configuração do worker inclui parâmetros como definições de conexão do Kafka, formato de serialização e frequência de commit de offset. Este exemplo baseia-se na documentação oficial do Kafka. Para mais informações, consulte 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=

Exemplo de arquivo de configuração do conector connect-tablestore-sink-quickstart.properties

A configuração do conector abrange parâmetros como a classe do conector, conexão com o Tablestore e mapeamento de dados. Para mais informações, consulte Configuração.

# 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

Acesse o diretório $KAFKA_HOME e execute o comando a seguir para iniciar o conector no modo standalone.

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

Etapa 3: Produzir novos registros

Acesse o diretório $KAFKA_HOME e execute o comando a seguir para iniciar um producer no console.

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

Consulte a tabela a seguir para obter descrições dos parâmetros.

Item de configuração	Exemplo	Descrição
--broker-list	localhost:9092	Endereço e porta do broker do cluster Kafka.
--topic	test	Nome do tópico. Por padrão, o tópico é criado automaticamente quando o Tablestore Sink Connector é iniciado. Você também pode criar o tópico manualmente.

Grave novas mensagens no tópico test.
Importante
Para importar dados para uma tabela de séries temporais, grave os dados no tópico em formato JSON.
```
{"m":"cpu","d":"127.0.0.1","region":"shanghai","level":1,"timestamp":1638868699090,"io":5.5,"cpu":"3.5"}
```
Faça login no console do Tablestore para visualizar os dados.

Na página Time Series Table Management, selecione a aba Data Management. Localize a série temporal com o nome de measurement cpu, a fonte de dados 127.0.0.1 e as tags level=1 e region=shanghai. Clique em Query Data. A consulta retorna um registro em que o horário é 2021-12-07 17:18:19, cpu é 3.5 e io é 5.5.