Descrição da configuração - Tablestore: Central de Documentação do Alibaba Cloud

Antes de iniciar o Tablestore Sink Connector, especifique pares de chave-valor para passar parâmetros ao processo do Kafka Connect. Este tópico fornece exemplos de configuração e descrições de parâmetros para mostrar como configurar o Tablestore Sink Connector.

Exemplos de configuração

Os itens de configuração variam de acordo com a sincronização de dados do Kafka para uma tabela de dados ou uma tabela de séries temporais no Tablestore. Os exemplos de configuração dos arquivos de configuração variam conforme o modo de operação. Esta seção fornece um exemplo de configuração da sincronização de dados do Kafka para uma tabela de dados no Tablestore. Para sincronizar dados com uma tabela de séries temporais no Tablestore, adicione itens de configuração específicos para a sincronização de dados do Kafka para uma tabela de séries temporais no Tablestore.

Modo autônomo

O código de exemplo a seguir mostra como configurar o arquivo de configuração no formato .properties para o Tablestore Sink Connector no modo autônomo:

# 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

Modo distribuído

O código de exemplo a seguir mostra como configurar o arquivo de configuração no formato .json para o Tablestore Sink Connector no modo distribuído:

{
  "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",
  }

Parâmetros

As tabelas a seguir descrevem os parâmetros do arquivo de configuração. Configure os parâmetros relacionados a séries temporais somente ao sincronizar dados do Kafka para uma tabela de séries temporais no Tablestore.

Parâmetros do Kafka Connect

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
name	string	Sim	tablestore-sink	O nome do conector. O nome do conector deve ser exclusivo.
connector.class	class	Sim	TableStoreSinkConnector	A classe Java do conector. Para usar o conector, especifique a classe do conector por meio de connector.class. Defina connector.class como o nome completo ou alias da classe do conector. O nome completo da classe é com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector e o alias é TableStoreSinkConnector. Exemplo: `connector.class=com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector`
tasks.max	integer	Sim	3	O número máximo de tarefas que podem ser criadas para o conector. Se não for possível criar o número máximo de tarefas, menos tarefas poderão ser criadas.
key.converter	string	Não	org.apache.kafka.connect.json.JsonConverter	O conversor de chaves usado para substituir o conversor de chaves padrão especificado no arquivo de configuração do worker.
value.converter	string	Não	org.apache.kafka.connect.json.JsonConverter	O conversor de valores usado para substituir o conversor de valores padrão especificado no arquivo de configuração do worker.
topics	list	Sim	test	A lista de tópicos do Kafka que podem ser especificados para o conector. Separe vários tópicos do Kafka com vírgulas (,). Especifique topics para gerenciar os tópicos definidos para o conector.

Parâmetros de conexão do conector

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
tablestore.endpoint	string	Sim	https://xxx.xxx.ots.aliyuncs.com	O endpoint da instância do Tablestore. Para obter mais informações, consulte Endpoints.
tablestore.mode	string	Sim	timeseries	O tipo da tabela de destino. Valor padrão: normal. Valores válidos: normal: uma tabela de dados no Tablestore. timeseries: uma tabela de séries temporais no Tablestore.
tablestore.access.key.id	string	Sim	LTAn********************	O AccessKey ID e o AccessKey secret da sua conta. Para obter mais informações sobre como obter o AccessKey ID e o AccessKey secret, consulte Criar um par de AccessKey.
tablestore.access.key.secret	string	Sim	zbnK**************************
tablestore.auth.mode	string	Sim	aksk	O modo de autenticação. Valor padrão: aksk. Valores válidos: aksk: usa o AccessKey ID e o AccessKey secret de uma conta do Alibaba Cloud ou de um usuário RAM para autenticação. Neste tópico, tablestore.auth.mode é definido como aksk. sts: usa as credenciais de acesso temporárias obtidas do Security Token Service (STS) para autenticação. Se o Tablestore estiver conectado ao Message Queue for Apache Kafka, defina tablestore.auth.mode como sts. Este é o valor padrão.
tablestore.instance.name	string	Sim	myotstest	O nome da instância do Tablestore.

Parâmetros de mapeamento de dados do conector

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
event.parse.class	class	Sim	DefaultEventParser	A classe Java do EventParser. Valor padrão: DefaultEventParser. O parser analisa registros de mensagens do Kafka para obter a coluna de chave primária e a coluna de atributo da tabela de dados. null O Tablestore impõe limites ao tamanho dos valores das colunas. Os valores das colunas de chave primária do tipo string ou binary não podem exceder 1 KB, e os valores das colunas de atributo não podem exceder 2 MB. Para obter mais informações, consulte Limites. Se os valores das colunas excederem os limites após a conversão dos tipos de dados, os registros de mensagens do Kafka serão processados como dados sujos. Para usar o DefaultEventParser, as chaves ou valores dos registros de mensagens do Kafka devem ser da classe Struct ou Map do Kafka Connect. Os campos selecionados em Struct devem ser de tipos de dados compatíveis com o Tablestore Sink Connector. Os campos são convertidos para tipos de dados do Tablestore com base na tabela de mapeamento de tipos de dados e gravados na tabela de dados. Os tipos de dados dos valores em Map devem ser compatíveis com o Tablestore Sink Connector. O Tablestore Sink Connector aceita os mesmos tipos de dados em Struct e Map. Os valores em Map são convertidos para dados do tipo binary e gravados na tabela de dados. Se os tipos de dados dos registros de mensagens do Kafka forem incompatíveis com o Tablestore Sink Connector, chame a operação definida por com.aliyun.tablestore.kafka.connect.parsers.EventParser para configurar o parser.
table.name.format	string	Não	kafka_<topic>	A string de formato para o nome da tabela de dados de destino do Tablestore. Valor padrão: <topic>. <topic> pode ser usado na string como espaço reservado para o tópico do qual os dados serão exportados. Por exemplo, se table.name.format for definido como kafka_<topic> e o nome do tópico do Kafka do qual os dados serão exportados for test, os registros de mensagens do Kafka do tópico test serão mapeados para a tabela chamada kafka_test no Tablestore. O parâmetro topics.assign.tables tem prioridade sobre table.name.format. Se topics.assign.tables for especificado, a configuração de table.name.format será ignorada.
topics.assign.tables	list	Sim	test:destTable	Especifica o mapeamento entre o tópico e a tabela de destino do Tablestore no formato `<topic_1>:<tablename_1>,<topic_2>:<tablename_2>` . Separe vários mapeamentos com vírgulas (,). Por exemplo, test:destTable especifica que os registros de mensagens do tópico chamado test são gravados na tabela de dados chamada destTable. O parâmetro topics.assign.tables tem prioridade sobre table.name.format. Se topics.assign.tables for especificado, a configuração de table.name.format será ignorada.
primarykey.mode	string	Não	kafka	O modo de chave primária da tabela de dados. Valores válidos: kafka: <connect_topic>_<connect_partition> e <connect_offset> são usados como chave primária da tabela de dados. O tópico do Kafka <connect_topic> e a partição <connect_partition> são separados por sublinhado (_), e <connect_offset> especifica o deslocamento do registro de mensagem na partição. record_key: os campos da classe Struct ou as chaves da classe Map nas chaves do registro são usados como chave primária da tabela de dados. record_value: os campos da classe Struct ou as chaves da classe Map nos valores do registro são usados como chave primária da tabela de dados. Configure este parâmetro junto com tablestore.<tablename>.primarykey.name e tablestore.<tablename>.primarykey.type. O valor deste parâmetro não diferencia maiúsculas de minúsculas.
tablestore.<tablename>.primarykey.name	list	Não	A,B	O nome da coluna de chave primária da tabela de dados. <tablename> é um espaço reservado para o nome da tabela de dados. O valor deste parâmetro contém de um a quatro nomes de colunas de chave primária separados por vírgulas (,). O nome da coluna de chave primária varia de acordo com o modo de chave primária. Se o modo de chave primária for definido como kafka, o valor padrão deste parâmetro é `topic_partition,offset`. No modo de chave primária kafka, não é necessário especificar os nomes das colunas de chave primária. Mesmo que sejam especificados, os nomes padrão das colunas de chave primária terão precedência. Se o modo de chave primária for definido como record_key, os campos da classe Struct ou as chaves da classe Map com os mesmos nomes das colunas de chave primária especificadas serão extraídos das chaves do registro como chave primária da tabela de dados. No modo de chave primária record_key, é necessário especificar os nomes das colunas de chave primária. Se o modo de chave primária for definido como record_value, os campos da classe Struct ou as chaves da classe Map com os mesmos nomes das colunas de chave primária especificadas serão extraídos dos valores do registro como chave primária da tabela de dados. No modo de chave primária record_value, é necessário especificar os nomes das colunas de chave primária. As colunas de chave primária da tabela de dados do Tablestore são sequenciais. Observe a sequência das colunas de chave primária ao definir tablestore.<tablename>.primarykey.name. Por exemplo, PRIMARY KEY (A, B, C) e PRIMARY KEY (A, C, B) possuem esquemas diferentes.
tablestore.<tablename>.primarykey.type	list	Não	string, integer	O tipo de dados da coluna de chave primária na tabela de dados. <tablename> é um espaço reservado para o nome da tabela de dados. O valor deste parâmetro contém de um a quatro tipos de dados das colunas de chave primária. Separe os tipos de dados das colunas de chave primária com vírgulas (,). A sequência dos tipos de dados das colunas de chave primária deve corresponder à sequência dos nomes das colunas de chave primária especificados por tablestore.<tablename>.primarykey.name. O valor deste parâmetro não diferencia maiúsculas de minúsculas. Valores válidos: integer, string, binary e auto_increment. O tipo de dados da coluna de chave primária varia de acordo com o modo de chave primária. Se o modo de chave primária for definido como kafka, o valor padrão deste parâmetro é `string, integer`. No modo de chave primária kafka, não é necessário especificar os tipos de dados das colunas de chave primária. Mesmo que sejam especificados, os tipos de dados padrão das colunas de chave primária terão precedência. Se o modo de chave primária for definido como record_key ou record_value, é necessário especificar os tipos de dados das colunas de chave primária. Se o tipo de dados especificado da coluna de chave primária entrar em conflito com o tipo de dados definido no esquema do Kafka, ocorrerá um erro de análise. Nesse caso, configure os parâmetros de Runtime Error para corrigir o erro. Se este parâmetro for definido como auto_increment, o campo dos registros de mensagens do Kafka será inserido na tabela de dados como uma coluna de chave primária de autoincremento ao gravar dados na tabela de dados.
tablestore.<tablename>.columns.whitelist.name	list	Não	A,B	O nome da coluna de atributo na lista de permissões de colunas de atributo. <tablename> é um espaço reservado para o nome da tabela de dados. Separe os nomes das colunas de atributo com vírgulas (,). Se este parâmetro não for configurado, todos os campos da classe Struct ou todas as chaves da classe Map nos valores do registro serão usados como colunas de atributo da tabela de dados. Ao configurar este parâmetro, os campos nos valores do registro serão filtrados com base na lista de permissões de colunas de atributo especificada para obter as colunas de atributo necessárias.
tablestore.<tablename>.columns.whitelist.type	list	Não	string, integer	O tipo de dados da coluna de atributo na lista de permissões de colunas de atributo. `<tablename>` é um espaço reservado para o nome da tabela de dados. Separe os tipos de dados das colunas de atributo com vírgulas (,). A sequência dos tipos de dados das colunas de atributo deve corresponder à sequência dos nomes das colunas de atributo especificados por `tablestore.<tablename>.columns.whitelist.name`. O valor deste parâmetro não diferencia maiúsculas de minúsculas. Valores válidos: integer, string, binary, boolean e double.

Parâmetros de gravação do conector

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
insert.mode	string	Não	put	O modo de gravação. Valor padrão: put. Valores válidos: put: os dados existentes são sobrescritos por uma linha de dados gravada na tabela. Este valor corresponde à operação PutRow do Tablestore. update: ao atualizar uma linha de dados, colunas de atributo são adicionadas à linha ou os valores das colunas de atributo existentes são atualizados. Este valor corresponde à operação UpdateRow do Tablestore. O valor deste parâmetro não diferencia maiúsculas de minúsculas.
insert.order.enable	boolean	Não	true	Especifica se os dados são gravados na tabela de dados na sequência em que são lidos. Valor padrão: true. Valores válidos: true: os registros de mensagens do Kafka são gravados na tabela de dados na sequência em que são lidos. false: os registros de mensagens do Kafka são gravados na tabela de dados sem uma sequência específica. Isso melhora o desempenho de gravação.
auto.create	boolean	Não	false	Especifica se a tabela de destino é criada automaticamente. Uma tabela de dados ou uma tabela de séries temporais pode ser criada automaticamente. Valor padrão: false. Valores válidos: true: o sistema cria automaticamente uma tabela de destino no Tablestore. false: o sistema não cria automaticamente uma tabela de destino no Tablestore.
delete.mode	string	Não	none	O modo de exclusão. A configuração deste parâmetro tem efeito somente quando os dados são sincronizados com uma tabela de dados e o modo de chave primária é definido como record_key. Valor padrão: none. Valores válidos: none: nenhuma operação de exclusão pode ser executada. row: linhas podem ser excluídas. Se um valor de registro estiver vazio, a linha correspondente será excluída. column: colunas de atributo podem ser excluídas. Se um valor de campo da classe Struct ou um valor de chave da classe Map nos valores do registro estiver vazio, a coluna de atributo correspondente será excluída. row_and_column: linhas e colunas de atributo podem ser excluídas. O valor deste parâmetro não diferencia maiúsculas de minúsculas. Este parâmetro é especificado com base no valor do parâmetro insert.mode. Para obter mais informações, consulte Apêndice: Sintaxe de exclusão.
buffer.size	integer	Não	1024	O número máximo de linhas que podem ser incluídas na fila de buffer na memória ao gravar dados na tabela de dados. Valor padrão: 1.024. O valor deste parâmetro deve ser uma potência de 2.
max.thread.count	integer	Não	3	O número de threads de callback usadas ao gravar dados na tabela de dados. Valor padrão = `Number of vCPUs + 1`.
max.concurrency	integer	Não	10	O número máximo de solicitações de gravação simultâneas que podem ser enviadas para gravar dados na tabela de dados. Valor padrão: 10.
bucket.count	integer	Não	3	O número de buckets nos quais os dados são gravados. Valor padrão: 3. Ao aumentar o valor deste parâmetro, a capacidade de gravação simultânea aumenta. No entanto, não é possível definir o valor deste parâmetro como maior que o número máximo de solicitações de gravação simultâneas especificado.
flush.Interval	integer	Não	10000	O intervalo de atualização da fila de buffer ao gravar dados na tabela de dados. Unidade: milissegundos. Valor padrão: 10.000.

Parâmetros de erro em tempo de execução do conector

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
runtime.error.tolerance	string	Não	none	A política de tratamento de erros utilizada quando ocorre um erro na análise ou gravação dos registros de mensagens do Kafka na tabela. Valor padrão: none. Valores válidos: none: um erro faz com que a tarefa de importação de dados do Tablestore Sink Connector falhe. all: os registros de mensagens com erros são ignorados e registrados em log. O valor deste parâmetro não diferencia maiúsculas de minúsculas.
runtime.error.mode	string	Não	ignore	Especifica como processar os registros de mensagens com erros relatados na análise ou gravação dos registros de mensagens do Kafka na tabela. Valor padrão: ignore. Valores válidos: ignore: todos os erros são ignorados. kafka: os registros de mensagens com erros e as mensagens de erro são armazenados em um tópico do Kafka diferente. Nesse caso, é necessário especificar runtime.error.bootstrap.servers e runtime.error.topic.name. As chaves e valores dos registros de mensagens com erros no novo tópico são os mesmos do tópico do qual os dados são exportados. O campo ErrorInfo é incluído no cabeçalho para registrar as mensagens de erro. tablestore: os registros de mensagens com erros e as mensagens de erro são armazenados em uma tabela de dados diferente do Tablestore. Nesse caso, é necessário especificar runtime.error.table.name. As colunas de chave primária da tabela de dados usada para registrar os registros de mensagens com erros e as mensagens de erro são `topic_partition (string type) and offset (integer type)`. As colunas de atributo da tabela de dados são key (tipo bytes), value (tipo bytes) e error_info (tipo string). Se runtime.error.mode for definido como kafka, é necessário serializar os cabeçalhos, chaves e valores dos registros de mensagens do Kafka. Se runtime.error.mode for definido como tablestore, é necessário serializar as chaves e valores dos registros de mensagens do Kafka. Por padrão, org.apache.kafka.connect.json.JsonConverter é usado para serializar dados e schemas.enable é definido como true. Use JsonConverter para desserializar dados e obter os dados originais. Para obter mais informações sobre o Converter, consulte Kafka Converter.
runtime.error.bootstrap.servers	string	Não	localhost:9092	O endereço do cluster do Kafka onde os registros de mensagens com erros e as mensagens de erro são armazenados.
runtime.error.topic.name	string	Não	errors	O nome do tópico do Kafka que armazena os registros de mensagens com erros e as mensagens de erro.
runtime.error.table.name	string	Não	errors	O nome da tabela do Tablestore que armazena os registros de mensagens com erros e as mensagens de erro.

Parâmetros relacionados a séries temporais

Parâmetro	Tipo	Obrigatório	Exemplo	Descrição
tablestore.timeseries.<tablename>.measurement	string	Sim	mName	Especifica que os valores correspondentes à chave especificada nos dados em formato JSON são gravados na tabela de séries temporais como valores do campo _m_name. Se tablestore.timeseries.<tablename>.measurement for definido como <topic>, os valores correspondentes à chave topic dos registros de mensagens do Kafka serão gravados na tabela de séries temporais como valores do campo _m_name. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades. Por exemplo, se o nome da tabela de séries temporais for test, o nome do parâmetro é tablestore.timeseries.test.measurement.
tablestore.timeseries.<tablename>.dataSource	string	Sim	ds	Especifica que os valores correspondentes à chave ds nos dados em formato JSON são gravados na tabela de séries temporais como valores do campo _data_source. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.<tablename>.tags	list	Sim	region,level	Especifica que os valores correspondentes às chaves region e level nos dados em formato JSON são gravados na tabela de séries temporais como valores do campo tags. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.<tablename>.time	string	Sim	timestamp	Especifica que os valores correspondentes à chave timestamp nos dados em formato JSON são gravados na tabela de séries temporais como valores do campo _time. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.<tablename>.time.unit	string	Sim	MILLISECONDS	A unidade dos valores do parâmetro tablestore.timeseries.<tablename>.time. Valores válidos: SECONDS, MILLISECONDS, MICROSECONDS e NANOSECONDS. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.<tablename>.field.name	list	Não	cpu,io	Especifica que as chaves cpu e io nos dados em formato JSON são gravadas na tabela de séries temporais como nomes de _field_name, e os valores correspondentes às chaves cpu e io nos dados em formato JSON são gravados na tabela de séries temporais como valores de _field_name. <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.<tablename>.field.type	string	Não	double,integer	O tipo de dados do campo especificado por tablestore.timeseries.<tablename>.field.name. Valores válidos: double, integer, string, binary e boolean. Separe vários tipos de dados com vírgulas (,). <tablename> no parâmetro é um espaço reservado para o nome da tabela de séries temporais. Modifique o nome do parâmetro conforme suas necessidades.
tablestore.timeseries.mapAll	boolean	Não	false	Especifica se os campos nos dados em formato JSON, além dos campos de chave primária e de tempo, são gravados na tabela de séries temporais como campos. Se tablestore.timeseries.mapAll for definido como false, é necessário configurar os parâmetros tablestore.timeseries.<tablename>.field.name e tablestore.timeseries.<tablename>.field.type.
tablestore.timeseries.toLowerCase	boolean	Não	true	Especifica se as chaves nos campos são convertidas para letras minúsculas antes de serem gravadas na tabela de séries temporais. As chaves nos campos são chaves em campos que não são de chave primária ou de tempo, ou chaves especificadas em tablestore.timeseries.<tablename>.field.name.
tablestore.timeseries.rowsPerBatch	integer	Não	50	O número máximo de linhas que podem ser gravadas no Tablestore em uma solicitação. Os valores máximo e padrão são 200.

Apêndice: Mapeamentos de tipos de dados entre Kafka e Tablestore

A tabela a seguir descreve os mapeamentos entre os tipos de dados do Kafka e do Tablestore.

Tipo de esquema do Kafka	Tipo de dados do Tablestore
STRING	STRING
INT8, INT16, INT32 e INT64	INTEGER
FLOAT32 e FLOAT64	DOUBLE
BOOLEAN	BOOLEAN
BYTES	BINARY

Apêndice: Sintaxe de exclusão

null

Este recurso é compatível somente quando os dados são sincronizados do Kafka para uma tabela de dados no Tablestore.

A tabela a seguir descreve os métodos usados para gravar dados em uma tabela de dados do Tablestore com base nas configurações do modo de gravação (insert.mode) e do modo de exclusão (delete.mode) quando os registros de mensagens contêm valores vazios e os dados são sincronizados do Kafka para uma tabela de dados no Tablestore.

insert.mode	put				update
delete.mode	none	row	column	row_and_column	none	row	column	row_and_column
Valores vazios	Sobrescrever	Excluir linhas	Sobrescrever	Excluir linhas	Dados sujos	Excluir linhas	Dados sujos	Excluir linhas
Todos os campos vazios nos valores	Sobrescrever	Sobrescrever	Sobrescrever	Sobrescrever	Dados sujos	Dados sujos	Excluir colunas	Excluir colunas
Alguns campos vazios nos valores	Sobrescrever	Sobrescrever	Sobrescrever	Sobrescrever	Ignorar valores vazios	Ignorar valores vazios	Excluir colunas	Excluir colunas