Descripción de la configuración - Tablestore - Centro de documentación de Alibaba Cloud

Antes de iniciar Tablestore Sink Connector, especifique pares clave-valor para pasar parámetros al proceso de Kafka Connect. Este tema proporciona ejemplos de configuración y descripciones de parámetros para mostrar cómo configurar Tablestore Sink Connector.

Ejemplos de configuración

Los elementos de configuración varían según si los datos se sincronizan desde Kafka hacia una tabla de datos o una tabla de series temporales en Tablestore. Los ejemplos de los archivos de configuración difieren según el modo de trabajo. Esta sección ofrece un ejemplo sobre cómo configurar la sincronización de datos desde Kafka hacia una tabla de datos en Tablestore. Para sincronizar datos con una tabla de series temporales en Tablestore, agregue los elementos de configuración específicos para la sincronización desde Kafka hacia este tipo de tabla.

Modo independiente

El siguiente código de ejemplo muestra cómo configurar el archivo en formato .properties para Tablestore Sink Connector en modo independiente:

# 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 distribuido

El siguiente código de ejemplo muestra cómo configurar el archivo en formato .json para Tablestore Sink Connector en modo distribuido:

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

Las siguientes tablas describen los parámetros del archivo de configuración. Solo configure los parámetros relacionados con series temporales cuando sincronice datos desde Kafka hacia una tabla de series temporales en Tablestore.

Parámetros de Kafka Connect

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
name	string	Sí	tablestore-sink	Nombre del conector. Debe ser único.
connector.class	class	Sí	TableStoreSinkConnector	Clase Java del conector. Para usar el conector, especifique su clase mediante connector.class. Establezca connector.class con el nombre completo o el alias de la clase del conector. El nombre completo es com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector y el alias es TableStoreSinkConnector. Por ejemplo: `connector.class=com.aliyun.tablestore.kafka.connect.TableStoreSinkConnector`
tasks.max	integer	Sí	3	Número máximo de tareas que se pueden crear para el conector. Si no se logra crear el número máximo de tareas, es posible que se creen menos.
key.converter	string	No	org.apache.kafka.connect.json.JsonConverter	Convertidor de claves que reemplaza al convertidor predeterminado especificado en el archivo de configuración del worker.
value.converter	string	No	org.apache.kafka.connect.json.JsonConverter	Convertidor de valores que reemplaza al convertidor predeterminado especificado en el archivo de configuración del worker.
topics	list	Sí	test	Lista de temas de Kafka que se pueden especificar para el conector. Separe varios temas con comas (,). Especifique temas para gestionar los que están asignados al conector.

Parámetros de conexión del conector

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
tablestore.endpoint	string	Sí	https://xxx.xxx.ots.aliyuncs.com	Endpoint de la instancia de Tablestore. Para más información, consulte Endpoints.
tablestore.mode	string	Sí	timeseries	Tipo de tabla de destino. Valor predeterminado: normal. Valores válidos: normal: tabla de datos en Tablestore. timeseries: tabla de series temporales en Tablestore.
tablestore.access.key.id	string	Sí	LTAn	AccessKey ID y AccessKey secret de su cuenta. Para obtener más información sobre cómo conseguirlos, consulte Create an AccessKey pair.
tablestore.access.key.secret	string	Sí	zbnK
tablestore.auth.mode	string	Sí	aksk	Modo de autenticación. Valor predeterminado: aksk. Valores válidos: aksk: utiliza el AccessKey ID y el AccessKey secret de una cuenta de Alibaba Cloud o un usuario RAM para la autenticación. En este tema, tablestore.auth.mode se establece en aksk. sts: utiliza las credenciales de acceso temporal obtenidas de Security Token Service (STS) para la autenticación. Si Tablestore se conecta a Message Queue for Apache Kafka, establezca tablestore.auth.mode en sts. Este es el valor predeterminado.
tablestore.instance.name	string	Sí	myotstest	Nombre de la instancia de Tablestore.

Parámetros de asignación de datos del conector

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
event.parse.class	class	Sí	DefaultEventParser	Clase Java del EventParser. Valor predeterminado: DefaultEventParser. El analizador procesa los registros de mensajes de Kafka para obtener la columna de clave primaria y la columna de atributo de la tabla de datos. Importante Tablestore impone límites en el tamaño de los valores de columna. Los valores de las columnas de clave primaria de tipo string o binary no pueden superar 1 KB, y los valores de las columnas de atributo no pueden exceder 2 MB. Para más información, consulte Limits. Si los valores de columna superan estos límites tras la conversión de tipos, los registros de mensajes de Kafka se tratan como datos erróneos. Para usar DefaultEventParser, las claves o valores de los registros de mensajes de Kafka deben pertenecer a la clase Struct o Map de Kafka Connect. Los campos seleccionados en Struct deben tener tipos de datos compatibles con Tablestore Sink Connector. Dichos campos se convierten a los tipos de datos de Tablestore según la tabla de asignación correspondiente y luego se escriben en la tabla de datos. Los tipos de datos de los valores en Map deben ser compatibles con Tablestore Sink Connector. Este conector admite los mismos tipos de datos en Struct y Map. Los valores en Map se convierten a tipo binary y después se escriben en la tabla de datos. Si los tipos de datos de los registros de mensajes de Kafka son incompatibles con Tablestore Sink Connector, invoque la operación definida por com.aliyun.tablestore.kafka.connect.parsers.EventParser para configurar el analizador.
table.name.format	string	No	kafka_<topic>	Cadena de formato para el nombre de la tabla de datos de destino en Tablestore. Valor predeterminado: <topic>. Use <topic> como marcador de posición para el tema del cual desea exportar datos. Por ejemplo, si table.name.format se establece en kafka_<topic> y el nombre del tema de Kafka es test, los registros de mensajes de Kafka de ese tema se asignarán a la tabla llamada kafka_test en Tablestore. topics.assign.tables tiene mayor prioridad que table.name.format. Si se especifica topics.assign.tables, se ignora la configuración de table.name.format.
topics.assign.tables	list	Sí	test:destTable	Especifica la asignación entre el tema y la tabla de destino en Tablestore con el formato `<topic_1>:<tablename_1>,<topic_2>:<tablename_2>`. Separe varias asignaciones con comas (,). Por ejemplo, test:destTable indica que los registros del tema llamado test se escriben en la tabla de datos llamada destTable. topics.assign.tables tiene mayor prioridad que table.name.format. Si se especifica topics.assign.tables, se ignora la configuración de table.name.format.
primarykey.mode	string	No	kafka	Modo de clave primaria de la tabla de datos. Valores válidos: kafka: se usan <connect_topic>_<connect_partition> y <connect_offset> como clave primaria de la tabla. El tema de Kafka <connect_topic> y la partición <connect_partition> se separan con un guion bajo (_), mientras que <connect_offset> indica el desplazamiento del registro dentro de la partición. record_key: se utilizan los campos de la clase Struct o las claves de la clase Map presentes en las claves del registro como clave primaria de la tabla. record_value: se utilizan los campos de la clase Struct o las claves de la clase Map presentes en los valores del registro como clave primaria de la tabla. Configure este parámetro junto con tablestore.<tablename>.primarykey.name y tablestore.<tablename>.primarykey.type. El valor de este parámetro no distingue entre mayúsculas y minúsculas.
tablestore.<tablename>.primarykey.name	list	No	A,B	Nombre de la columna de clave primaria de la tabla. <tablename> es un marcador de posición para el nombre de la tabla. El valor contiene de uno a cuatro nombres de columnas de clave primaria separados por comas (,). El nombre de la columna de clave primaria varía según el modo seleccionado. Si el modo es kafka, el valor predeterminado es `topic_partition,offset`. En este modo no es necesario especificar los nombres; incluso si se definen, prevalecen los predeterminados. Si el modo es record_key, se extraen de las claves del registro los campos de la clase Struct o las claves de la clase Map que coincidan con los nombres especificados para formar la clave primaria. En este caso, es obligatorio definir los nombres. Si el modo es record_value, se extraen de los valores del registro los campos de la clase Struct o las claves de la clase Map que coincidan con los nombres especificados para formar la clave primaria. Aquí también es obligatorio definir los nombres. Las columnas de clave primaria en Tablestore son secuenciales. Considere el orden al definir tablestore.<tablename>.primarykey.name. Por ejemplo, PRIMARY KEY (A, B, C) y PRIMARY KEY (A, C, B) generan esquemas distintos.
tablestore.<tablename>.primarykey.type	list	No	string, integer	Tipo de dato de la columna de clave primaria. <tablename> es un marcador de posición para el nombre de la tabla. El valor incluye de uno a cuatro tipos separados por comas (,). El orden debe coincidir con el definido en tablestore.<tablename>.primarykey.name. No distingue entre mayúsculas y minúsculas. Valores válidos: integer, string, binary y auto_increment. El tipo de dato depende del modo de clave primaria. En modo kafka, el valor predeterminado es `string, integer`. No es necesario especificar tipos en este modo; si se definen, se ignoran y se usan los predeterminados. En modos record_key o record_value, es obligatorio especificar los tipos de dato. Si el tipo especificado entra en conflicto con el esquema de Kafka, ocurrirá un error de análisis. Configure los parámetros de Runtime Error para manejarlo. Si se establece en auto_increment, el campo del registro se inserta como columna de clave primaria autoincrementable al escribir en la tabla.
tablestore.<tablename>.columns.whitelist.name	list	No	A,B	Nombre de la columna de atributo incluida en la lista blanca. <tablename> es un marcador de posición. Separe los nombres con comas (,). Si no configura este parámetro, todos los campos de la clase Struct o todas las claves de la clase Map en los valores del registro se usan como columnas de atributo. Si lo configura, se filtran los campos según la lista blanca para obtener solo las columnas deseadas.
tablestore.<tablename>.columns.whitelist.type	list	No	string, integer	Tipo de dato de la columna de atributo en la lista blanca. `<tablename>` es un marcador de posición. Separe los tipos con comas (,). El orden debe coincidir con el definido en `tablestore.<tablename>.columns.whitelist.name`. No distingue entre mayúsculas y minúsculas. Valores válidos: integer, string, binary, boolean y double.

Parámetros de escritura del conector

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
insert.mode	string	No	put	Modo de escritura. Valor predeterminado: put. Valores válidos: put: los datos existentes se sobrescriben con la nueva fila escrita. Corresponde a la operación PutRow de Tablestore. update: al actualizar una fila, se agregan columnas de atributo o se modifican las existentes. Corresponde a la operación UpdateRow de Tablestore. Este valor no distingue entre mayúsculas y minúsculas.
insert.order.enable	boolean	No	true	Indica si los datos se escriben respetando el orden de lectura. Valor predeterminado: true. Valores válidos: true: los registros se escriben siguiendo el orden en que fueron leídos. false: los registros se escriben sin un orden específico, lo cual mejora el rendimiento de escritura.
auto.create	boolean	No	false	Indica si se debe crear automáticamente la tabla de destino (ya sea de datos o de series temporales). Valor predeterminado: false. Valores válidos: true: el sistema crea automáticamente la tabla en Tablestore. false: el sistema no crea la tabla automáticamente.
delete.mode	string	No	none	Modo de eliminación. Solo aplica cuando se sincroniza hacia una tabla de datos y el modo de clave primaria es record_key. Valor predeterminado: none. Valores válidos: none: no se realizan operaciones de eliminación. row: permite eliminar filas. Si el valor del registro está vacío, se elimina la fila correspondiente. column: permite eliminar columnas de atributo. Si un campo de la clase Struct o una clave de la clase Map en los valores del registro está vacío, se elimina dicha columna. row_and_column: permite eliminar tanto filas como columnas de atributo. Este valor no distingue entre mayúsculas y minúsculas. Se define en función del parámetro insert.mode. Para más información, consulte Appendix: Delete syntax.
buffer.size	integer	No	1024	Número máximo de filas permitidas en la cola de búfer en memoria durante la escritura. Valor predeterminado: 1024. Debe ser una potencia de 2.
max.thread.count	integer	No	3	Cantidad de hilos de devolución de llamada usados durante la escritura. Valor predeterminado = `Number of vCPUs + 1`.
max.concurrency	integer	No	10	Número máximo de solicitudes de escritura simultáneas permitidas. Valor predeterminado: 10.
bucket.count	integer	No	3	Número de buckets donde se escriben los datos. Valor predeterminado: 3. Aumentar este valor incrementa la capacidad de escritura concurrente, pero no puede superar el límite definido en max.concurrency.
flush.Interval	integer	No	10000	Intervalo de vaciado de la cola de búfer durante la escritura. Unidad: milisegundos. Valor predeterminado: 10000.

Parámetros de Runtime Error del conector

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
runtime.error.tolerance	string	No	none	Política de manejo de errores durante el análisis o escritura de registros. Valor predeterminado: none. Valores válidos: none: cualquier error provoca el fallo de la tarea de importación. all: se omiten y registran los registros que presentan errores. No distingue entre mayúsculas y minúsculas.
runtime.error.mode	string	No	ignore	Define cómo procesar los registros con errores durante el análisis o escritura. Valor predeterminado: ignore. Valores válidos: ignore: se ignoran todos los errores. kafka: los registros fallidos y sus mensajes de error se almacenan en otro tema de Kafka. Requiere configurar runtime.error.bootstrap.servers y runtime.error.topic.name. Las claves y valores en el nuevo tema son idénticos a los originales, y se añade el campo ErrorInfo en el encabezado para registrar el error. tablestore: los registros fallidos y sus mensajes de error se guardan en otra tabla de datos de Tablestore. Requiere configurar runtime.error.table.name. Las columnas de clave primaria de esta tabla son `topic_partition (tipo string) y offset (tipo integer)`. Las columnas de atributo son key (tipo bytes), value (tipo bytes) y error_info (tipo string). Si runtime.error.mode es kafka, serialice encabezados, claves y valores. Si es tablestore, solo claves y valores. Por defecto se usa org.apache.kafka.connect.json.JsonConverter con schemas.enable en true. Use JsonConverter para deserializar y recuperar los datos originales. Para más detalles, consulte Kafka Converter.
runtime.error.bootstrap.servers	string	No	localhost:9092	Dirección del clúster de Kafka donde se almacenarán los registros fallidos y los mensajes de error.
runtime.error.topic.name	string	No	errors	Nombre del tema de Kafka destinado a almacenar los registros fallidos y los mensajes de error.
runtime.error.table.name	string	No	errors	Nombre de la tabla de Tablestore destinada a almacenar los registros fallidos y los mensajes de error.

Parámetros relacionados con series temporales

Parámetro	Tipo	Obligatorio	Ejemplo	Descripción
tablestore.timeseries.<tablename>.measurement	string	Sí	mName	Indica que los valores asociados a la clave especificada en datos JSON se escriben en la tabla de series temporales como valores del campo _m_name. Si tablestore.timeseries.<tablename>.measurement se establece en <topic>, los valores de la clave topic de los registros de Kafka se escribirán en el campo _m_name. <tablename> es un marcador de posición para el nombre de la tabla. Adapte el nombre del parámetro según sus necesidades. Por ejemplo, si la tabla se llama test, el parámetro sería tablestore.timeseries.test.measurement.
tablestore.timeseries.<tablename>.dataSource	string	Sí	ds	Indica que los valores asociados a la clave ds en datos JSON se escriben en la tabla de series temporales como valores del campo _data_source. <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.<tablename>.tags	list	Sí	region,level	Indica que los valores asociados a las claves region y level en datos JSON se escriben en la tabla de series temporales como valores del campo tags. <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.<tablename>.time	string	Sí	timestamp	Indica que los valores asociados a la clave timestamp en datos JSON se escriben en la tabla de series temporales como valores del campo _time. <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.<tablename>.time.unit	string	Sí	MILLISECONDS	Unidad de tiempo para el parámetro tablestore.timeseries.<tablename>.time. Valores válidos: SECONDS, MILLISECONDS, MICROSECONDS y NANOSECONDS. <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.<tablename>.field.name	list	No	cpu,io	Indica que las claves cpu e io en datos JSON se escriben como nombres de _field_name, y sus valores correspondientes se escriben como valores de _field_name en la tabla de series temporales. <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.<tablename>.field.type	string	No	double,integer	Tipo de dato del campo definido en tablestore.timeseries.<tablename>.field.name. Valores válidos: double, integer, string, binary y boolean. Separe varios tipos con comas (,). <tablename> es un marcador de posición. Adapte el nombre del parámetro según sus necesidades.
tablestore.timeseries.mapAll	boolean	No	false	Indica si los campos del JSON (distintos a los de clave primaria y tiempo) se escriben directamente como campos en la tabla de series temporales. Si tablestore.timeseries.mapAll es false, configure obligatoriamente tablestore.timeseries.<tablename>.field.name y tablestore.timeseries.<tablename>.field.type.
tablestore.timeseries.toLowerCase	boolean	No	true	Indica si las claves de los campos se convierten a minúsculas antes de escribirse en la tabla. Aplica a claves de campos no primarios, no temporales o aquellas definidas en tablestore.timeseries.<tablename>.field.name.
tablestore.timeseries.rowsPerBatch	integer	No	50	Número máximo de filas que se pueden escribir en Tablestore por solicitud. El valor máximo y predeterminado es 200.

Apéndice: Asignación de tipos de datos entre Kafka y Tablestore

La siguiente tabla describe las equivalencias entre los tipos de datos de Kafka y Tablestore.

Tipo de esquema de Kafka	Tipo de dato en Tablestore
STRING	STRING
INT8, INT16, INT32 y INT64	INTEGER
FLOAT32 y FLOAT64	DOUBLE
BOOLEAN	BOOLEAN
BYTES	BINARY

Apéndice: Sintaxis de eliminación

Nota

Esta función solo es compatible cuando se sincronizan datos desde Kafka hacia una tabla de datos en Tablestore.

La siguiente tabla describe cómo se escriben los datos en una tabla de Tablestore según la configuración del modo de escritura (insert.mode) y el modo de eliminación (delete.mode), considerando registros con valores vacíos durante la sincronización desde Kafka.

insert.mode	put				update
delete.mode	none	row	column	row_and_column	none	row	column	row_and_column
Valores vacíos	Sobrescribir	Eliminar filas	Sobrescribir	Eliminar filas	Datos erróneos	Eliminar filas	Datos erróneos	Eliminar filas
Todos los campos vacíos en valores	Sobrescribir	Sobrescribir	Sobrescribir	Sobrescribir	Datos erróneos	Datos erróneos	Eliminar columnas	Eliminar columnas
Algunos campos vacíos en valores	Sobrescribir	Sobrescribir	Sobrescribir	Sobrescribir	Ignorar valores vacíos	Ignorar valores vacíos	Eliminar columnas	Eliminar columnas