Todos os produtos
Search
Central de documentação

Tablestore:Sincronizar dados do Kafka para uma tabela de dados

Última atualização: Jul 16, 2026

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.

    Nota

    Também é possível usar o Tablestore sink connector para criar automaticamente a tabela de dados de destino. Para isso, defina o parâmetro auto.create como true.

  • Obtenha um AccessKey. Para mais informações, consulte Criar um AccessKey.

Etapa 1: Implantar o Tablestore Sink Connector

  1. 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.

      1. Execute o comando a seguir para baixar o source do Tablestore sink connector.

        git clone https://github.com/aliyun/kafka-connect-tablestore.git
      2. Acesse o diretório do source baixado e execute o comando a seguir para empacotar o código usando o Maven.

        mvn clean package -DskipTests

        Apó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.

  2. 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

  1. 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
  2. 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

  1. 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=
  2. Acesse o diretório $KAFKA_HOME e execute o comando a seguir para iniciar o connector no modo distribuído.

    Importante

    Inicie o processo do worker em cada nó.

    bin/connect-distributed.sh config/connect-distributed.properties
  3. Gerencie os connectors usando a rest API. Para mais informações, consulte rest API.

    1. 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"
        }
      }
    2. 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/connectors

      http://localhost:8083/connectors é o endereço do service rest do Kafka. Modifique este endereço conforme necessário.

Etapa 3: Produzir novos registros

  1. 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

    Para 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.

  2. 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
          }
      }
  3. 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=foo e quantity=100.