O Tablestore sink connector consulta o Kafka em busca de mensagens dos tópicos inscritos, analisa os registros das mensagens e importa os dados em lote para uma tabela de dados do Tablestore.
Pré-requisitos
O Kafka está instalado e tanto o ZooKeeper quanto o Kafka estão em execução. Para mais informações, consulte a documentação oficial do Kafka.
-
Ative o service Tablestore e crie uma instância e uma tabela de dados. Para mais informações, consulte Início rápido para modelos de tabela ampla.
NotaTambém é possível usar o Tablestore sink connector para criar automaticamente a tabela de dados de destino. Para isso, defina o parâmetro
auto.createcomotrue. Obtenha um AccessKey. Para mais informações, consulte Criar um AccessKey.
Etapa 1: Implantar o Tablestore Sink Connector
-
Obtenha o Tablestore sink connector de uma das seguintes maneiras.
-
Baixe e compile o código-fonte do GitHub. O source está disponível em Código-fonte do Tablestore Sink Connector.
-
Execute o comando a seguir para baixar o source do Tablestore sink connector.
git clone https://github.com/aliyun/kafka-connect-tablestore.git -
Acesse o diretório do source baixado e execute o comando a seguir para empacotar o código usando o Maven.
mvn clean package -DskipTestsApós a conclusão da compilação, o pacote gerado, como kafka-connect-tablestore-1.0.jar, será armazenado no diretório target.
-
Baixe o pacote kafka-connect-tablestore 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 suporta dois modos de execução: standalone e distribuído. Escolha o modo adequado às suas necessidades.
Modo Standalone
-
Modifique o arquivo de configuração do worker connect-standalone.properties e o arquivo de configuração do connector connect-tablestore-sink-quickstart.properties conforme necessário.
-
Exemplo do arquivo de configuração do worker connect-standalone.properties
A configuração do worker inclui parâmetros como configuraçõ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=true 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 connector: connect-tablestore-sink-quickstart.properties
A configuração do connector inclui parâmetros como a classe do connector, conexão com o Tablestore e mapeamento de dados. Para mais informações, consulte Configuração.
# Set the connector name. name=tablestore-sink # Specify the connector class. connector.class=TableStoreSinkConnector # Set the maximum number of tasks. tasks.max=1 # Specify the list of Kafka topics from which to export data. topics=test # The following are the Tablestore connection parameter settings. # The endpoint of the Tablestore instance. tablestore.endpoint=https://xxx.xxx.ots.aliyuncs.com # Your AccessKey ID and AccessKey secret. tablestore.access.key.id =xxx tablestore.access.key.secret=xxx # The Tablestore instance name. tablestore.instance.name=xxx # A format string for the destination Tablestore table name. <topic> is a placeholder for the original topic name. The default value is <topic>. # Examples: # table.name.format=kafka_<topic>, message records from the 'test' topic will be written to the 'kafka_test' data table. # table.name.format= # The primary key mode. The default value is kafka. # The primary key of the Tablestore data table will be composed of <topic>_<partition> (the Kafka topic and partition, separated by "_") and <offset> (the offset of the message record in the partition). # primarykey.mode= # Automatically create the destination table. The default value is false. auto.create=true
-
-
Acesse o diretório $KAFKA_HOME e execute o comando a seguir para iniciar o connector no modo standalone.
bin/connect-standalone.sh config/connect-standalone.properties config/connect-tablestore-sink-quickstart.properties
Modo distribuído
-
Modifique o arquivo de configuração do worker connect-distributed.properties conforme necessário.
A configuração do worker inclui parâmetros como configurações de conexão do Kafka, formato de serialização e frequência de commit de offset. Ela também especifica os tópicos que armazenam informações para cada connector. Crie esses tópicos antecipadamente. Este exemplo baseia-se na documentação oficial do Kafka. Para mais informações, consulte Kafka Connect.
offset.storage.topic: Um tópico compactado usado para armazenar os offsets de cada connector.
config.storage.topic: Um tópico compactado usado para armazenar configurações de connectors e tarefas. O número de partições para este tópico deve ser definido como 1.
status.storage.topic: Um tópico compactado usado para armazenar informações de status do 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. ## # This file contains some of the configurations for the Kafka Connect distributed worker. This file is intended # to be used with the examples, and some settings may differ from those used in a production system, especially # the `bootstrap.servers` and those specifying replication factors. # A list of host/port pairs to use for establishing the initial connection to the Kafka cluster. bootstrap.servers=localhost:9092 # unique name for the cluster, used in forming the Connect cluster group. Note that this must not conflict with consumer group IDs group.id=connect-cluster # 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=true # Topic to use for storing offsets. This topic should have many partitions and be replicated and compacted. # Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. # Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value. # Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able # to run this example on a single-broker cluster and so here we instead set the replication factor to 1. offset.storage.topic=connect-offsets offset.storage.replication.factor=1 #offset.storage.partitions=25 # Topic to use for storing connector and task configurations; note that this should be a single partition, highly replicated, # and compacted topic. Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. # Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value. # Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able # to run this example on a single-broker cluster and so here we instead set the replication factor to 1. config.storage.topic=connect-configs config.storage.replication.factor=1 # Topic to use for storing statuses. This topic can have multiple partitions and should be replicated and compacted. # Kafka Connect will attempt to create the topic automatically when needed, but you can always manually create # the topic before starting Kafka Connect if a specific topic configuration is needed. # Most users will want to use the built-in default replication factor of 3 or in some cases even specify a larger value. # Since this means there must be at least as many brokers as the maximum replication factor used, we'd like to be able # to run this example on a single-broker cluster and so here we instead set the replication factor to 1. status.storage.topic=connect-status status.storage.replication.factor=1 #status.storage.partitions=5 # Flush much faster than normal, which is useful for testing/debugging offset.flush.interval.ms=10000 # These are provided to inform the user about the presence of the REST host and port configs # Hostname & Port for the REST API to listen on. If this is set, it will bind to the interface used to listen to requests. #rest.host.name= #rest.port=8083 # The Hostname & Port that will be given out to other workers to connect to i.e. URLs that are routable from other servers. #rest.advertised.host.name= #rest.advertised.port= # 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 # Examples: # plugin.path=/usr/local/share/java,/usr/local/share/kafka/plugins,/opt/connectors, #plugin.path= -
Acesse o diretório $KAFKA_HOME e execute o comando a seguir para iniciar o connector no modo distribuído.
ImportanteInicie o processo do worker em cada nó.
bin/connect-distributed.sh config/connect-distributed.properties -
Gerencie os connectors usando a rest API. Para mais informações, consulte rest API.
-
Crie o arquivo connect-tablestore-sink-quickstart.json no caminho config e adicione o seguinte conteúdo de exemplo.
O arquivo de configuração do connector especifica parâmetros como pares chave-valor no formato JSON. Esses parâmetros incluem a classe do connector, configurações de conexão do Tablestore e mapeamento de dados. Para mais informações, consulte Configuração.
{ "name": "tablestore-sink", "config": { "connector.class":"TableStoreSinkConnector", "tasks.max":"1", "topics":"test", "tablestore.endpoint":"https://xxx.xxx.ots.aliyuncs.com", "tablestore.access.key.id":"xxx", "tablestore.access.key.secret":"xxx", "tablestore.instance.name":"xxx", "table.name.format":"<topic>", "primarykey.mode":"kafka", "auto.create":"true" } } -
Execute o comando a seguir para iniciar o Tablestore sink connector.
curl -i -k -H "Content-type: application/json" -X POST -d @config/connect-tablestore-sink-quickstart.json http://localhost:8083/connectorshttp://localhost:8083/connectorsé o endereço do service rest do Kafka. Modifique este endereço conforme necessário.
-
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 testPara obter uma descrição dos parâmetros, consulte a tabela a seguir.
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 inicia. Também é possível criar o tópico manualmente.
-
Escreva mensagens no tópico test.
-
Mensagem do tipo Struct
{ "schema":{ "type":"struct", "fields":[ { "type":"int32", "optional":false, "field":"id" }, { "type":"string", "optional":false, "field":"product" }, { "type":"int64", "optional":false, "field":"quantity" }, { "type":"double", "optional":false, "field":"price" } ], "optional":false, "name":"record" }, "payload":{ "id":1, "product":"foo", "quantity":100, "price":50 } } -
Mensagem do tipo Map
{ "schema":{ "type":"map", "keys":{ "type":"string", "optional":false }, "values":{ "type":"int32", "optional":false }, "optional":false }, "payload":{ "id":1 } }
-
-
Faça login no console do Tablestore para visualizar os dados.
Uma tabela de dados chamada test é criada automaticamente na instância do Tablestore. A tabela de resultados da consulta contém as seguintes colunas: topic_partition (chave primária), offset (chave primária), id, price, product e quantity. Nos dados de exemplo, o primeiro registro tem topic_partition=
test_3, offset=0, id=AAAAAQ==e os outros campos estão vazios. O segundo registro tem topic_partition=test_34, offset=0, id=1, price=50.0, product=fooe quantity=100.