En esta página, se explica cómo usar el conector de Kafka para consumir y reenviar datos de los flujos de cambios de Spanner.
Conceptos básicos
A continuación, se describen los conceptos principales del conector de Kafka.
Debezium
Debezium es un proyecto de código abierto que proporciona una plataforma de transmisión de datos de baja latencia para la captura de datos modificados.
Conector de Kafka
El conector Kafka proporciona una abstracción sobre la API de Spanner para publicar flujos de cambios de Spanner en Kafka, Con este conector, No tiene que administrar el ciclo de vida de la partición de flujos de cambios lo que es necesario cuando usas la API de Spanner directamente.
El conector de Kafka produce un evento de cambio para cada modificación de registro de cambio de datos y envía los registros de eventos de cambio a un tema de Kafka independiente para cada tabla con seguimiento de flujo de cambios. Un mod de registro de cambios de datos representa una sola modificación (inserción, actualización o eliminación) que se capturó. Un solo registro de cambio de datos puede contener más de un mod.
Salida del conector de Kafka
El conector de Kafka reenvía
los registros de flujos de cambios
en otro tema de Kafka. El nombre del tema de salida debe ser connector_name.table_name.
Si el tema no existe, el conector de Kafka crea automáticamente un tema con ese nombre.
También puedes configurar transformaciones de enrutamiento de temas para volver a enrutar los registros a los temas que especifiques. Si quieres usar el enrutamiento de temas, inhabilita la funcionalidad de marca de agua baja.
Orden de los registros
Los registros se ordenan por marca de tiempo de confirmación por clave primaria en
los temas de Kafka. Los registros que pertenecen a diferentes claves primarias no tienen
garantías de los pedidos. Los registros con la misma clave primaria se almacenan
misma partición del tema de Kafka. Si quieres procesar transacciones completas,
puedes usar el registro de cambios de datos
Los campos server_transaction_id y number_of_records_in_transaction para
ensamblar una transacción de Spanner.
Eventos de cambio
El conector de Kafka genera un evento de cambio de datos para cada INSERT, UPDATE,
y DELETE. Cada evento contiene una clave y valores para la fila modificada.
Puedes usar conversores de Kafka Connect para producir eventos de cambio de datos en formatos Protobuf, AVRO, JSON o JSON Schemaless. Si usas un
conversor de Kafka Connect que produce esquemas, el evento contiene
esquemas separados para la clave y los valores. De lo contrario, el evento solo contiene
la clave y los valores.
El esquema de la clave nunca cambia. El esquema de los valores es una combinación de todas las columnas de las que el flujo de cambios hizo un seguimiento desde la hora de inicio del conector.
Si configuras el conector para que produzca eventos JSON, el evento de cambio de salida contiene cinco campos:
El primer campo
schemaespecifica un esquema Kafka Connect que describe el esquema de claves de Spanner.El primer campo
payloadtiene la estructura descrita por el camposchemaanterior y contiene la clave de la fila que se cambió.El segundo campo
schemaespecifica el esquema de Kafka Connect que describe el esquema de la fila modificada.El segundo campo
payloadtiene la estructura descrita por el camposchemaanterior y contiene los datos reales de la fila que se modificó.El campo
sourcees un campo obligatorio que describe los metadatos de origen del evento.
El siguiente es un ejemplo de un evento de cambio de datos:
{
// The schema for the Spanner key.
"schema": {
"type": "struct",
"name": "customers.Key",
"optional": false,
"fields": [
{
"type": "int64",
"optional": "false"
"field": "false"
}
]
},
// The value of the Spanner key.
"payload": {
"id": "1"
},
// The schema for the payload, which contains the before and after values
// of the changed row. The schema for the payload contains all the
// columns that the change stream has tracked since the connector start
// time.
"schema": {
"type": "struct",
"fields": [
{
// The schema for the before values of the changed row.
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "id"
},
{
"type": "string",
"optional": true,
"field": "first_name"
}
],
"optional": true,
"name": "customers.Value",
"field": "before"
},
{
// The schema for the after values of the changed row.
"type": "struct",
"fields": [
{
"type": "int32",
"optional": false,
"field": "id"
},
{
"type": "string",
"optional": false,
"field": "first_name"
}
],
"optional": true,
"name": "customers.Value",
"field": "after"
},
{
// The schema for the source metadata for the event.
"type": "struct",
"fields": [
{
"type": "string",
"optional": false,
"field": "version"
},
{
"type": "string",
"optional": false,
"field": "connector"
},
{
"type": "string",
"optional": false,
"field": "name"
},
{
"type": "int64",
"optional": false,
"field": "ts_ms"
},
{
"type": "boolean",
"optional": true,
"default": false,
"field": "snapshot"
},
{
"type": "string",
"optional": false,
"field": "db"
},
{
"type": "string",
"optional": false,
"field": "sequence"
},
{
"type": "string",
"optional": false,
"field": "project_id"
},
{
"type": "string",
"optional": false,
"field": "instance_id"
},
{
"type": "string",
"optional": false,
"field": "database_id"
},
{
"type": "string",
"optional": false,
"field": "change_stream_name"
},
{
"type": "string",
"optional": true,
"field": "table"
}
{
"type": "string",
"optional": true,
"field": "server_transaction_id"
}
{
"type": "int64",
"optional": true,
"field": "low_watermark"
}
{
"type": "int64",
"optional": true,
"field": "read_at_timestamp"
}
{
"type": "int64",
"optional": true,
"field": "number_of_records_in_transaction"
}
{
"type": "string",
"optional": true,
"field": "transaction_tag"
}
{
"type": "boolean",
"optional": true,
"field": "system_transaction"
}
{
"type": "string",
"optional": true,
"field": "value_capture_type"
}
{
"type": "string",
"optional": true,
"field": "partition_token"
}
{
"type": "int32",
"optional": true,
"field": "mod_number"
}
{
"type": "boolean",
"optional": true,
"field": "is_last_record_in_transaction_in_partition"
}
{
"type": "int64",
"optional": true,
"field": "number_of_partitions_in_transaction"
}
],
"optional": false,
"name": "io.debezium.connector.spanner.Source",
"field": "source"
},
]
{
"type": "string",
"optional": false,
"field": "op"
},
{
"type": "int64",
"optional": true,
"field": "ts_ms"
}
],
"optional": false,
"name": "connector_name.customers.Envelope"
},
"payload": {
// The values of the row before the event.
"before": null,
// The values of the row after the event.
"after": {
"id": 1,
"first_name": "Anne",
}
},
// The source metadata.
"source": {
"version": "{debezium-version}",
"connector": "spanner",
"name": "spanner_connector",
"ts_ms": 1670955531785,
"snapshot": "false",
"db": "database",
"sequence": "1",
"project_id": "project",
"instance_id": "instance",
"database_id": "database",
"change_stream_name": "change_stream",
"table": "customers",
"server_transaction_id": "transaction_id",
"low_watermark": 1670955471635,
"read_at_timestamp": 1670955531791,
"number_records_in_transaction": 2,
"transaction_tag": "",
"system_transaction": false,
"value_capture_type": "OLD_AND_NEW_VALUES",
"partition_token": "partition_token",
"mod_number": 0,
"is_last_record_in_transaction_in_partition": true,
"number_of_partitions_in_transaction": 1
},
"op": "c",
"ts_ms": 1559033904863 //
}
Marca de agua baja
La marca de agua baja describe la hora T en la que se garantiza que el conector de Kafka transmitió y publicó en un tema de Kafka todos los eventos con marca de tiempo. T.
Puedes habilitar la marca de agua baja en el conector de Kafka con el parámetro gcp.spanner.low-watermark.enabled. Este parámetro está inhabilitado
de forma predeterminada. Si la marca de agua baja está habilitada, el campo low_watermark en el registro de cambios de datos del flujo de cambios se propaga con la marca de tiempo de la marca de agua baja actual del conector de Kafka.
Si no se producen registros, el conector de Kafka envía marca de agua de “latidos de corazón” a los temas de salida de Kafka que detectó el conector.
Estos latidos de marca de agua son registros vacíos, excepto por el
low_watermark. Luego, puedes usar la marca de agua baja para realizar agregaciones basadas en el tiempo.
Por ejemplo, puedes usar la marca de agua baja para ordenar los eventos por marca de tiempo de confirmación en las claves primarias.
Temas de metadatos
El conector de Kafka, al igual que el marco de trabajo Kafka Connect, crea varias de metadatos para almacenar información relacionada con los conectores. No se recomienda modificar la configuración ni el contenido de estos temas de metadatos.
Los siguientes son los temas de metadatos:
_consumer_offsets: Es un tema que Kafka crea automáticamente. Almacena compensaciones para los consumidores creadas en el conector de Kafka._kafka-connect-offsets: Es un tema que Kafka Connect crea automáticamente. Almacena los desplazamientos del conector._sync_topic_spanner_connector_connectorname: un tema creado automáticamente por el conector. Almacena metadatos relacionados con particiones de flujos de cambios._rebalancing_topic_spanner_connector_connectorname: un tema creado automáticamente por el conector. Se usa para determinar el estado de actividad de la tarea del conector._debezium-heartbeat.connectorname: Es un tema que se usa para procesar los mensajes de estado del flujo de cambios de Spanner.
Entorno de ejecución del conector de Kafka
A continuación, se describe el entorno de ejecución del conector de Kafka.
Escalabilidad
El conector de Kafka es escalable horizontalmente y se ejecuta en una o más tareas distribuidas entre varios trabajadores de Kafka Connect.
Garantías de entrega de mensajes
El conector de Kafka admite la garantía de entrega al menos una vez.
Tolerancia a errores
El conector de Kafka es tolerante a las fallas. A medida que el conector de Kafka lee los cambios y produce eventos, registra la última marca de tiempo de confirmación procesada para cada partición de flujo de cambios. Si el conector de Kafka se detiene por algún motivo (incluido fallas de comunicación, problemas de red o fallas de software), al reiniciar el conector de Kafka continúa transmitiendo registros donde se interrumpió.
El conector de Kafka lee el esquema de información en la marca de tiempo de inicio del conector de Kafka para recuperar la información del esquema. De forma predeterminada, Spanner no puede leer el esquema de información en marcas de tiempo de lectura anteriores al período de retención de la versión, que se establece de forma predeterminada en una hora. Si quieres iniciar el conector antes del una hora después, debes aumentar la retención de versiones de la base de datos período.
Configura el conector de Kafka
Crear transmisión de cambios
Si deseas obtener detalles para crear un flujo de cambios, consulta Cómo crear un flujo de cambios. Para continuar con los siguientes pasos, se requiere una instancia de Spanner con un flujo de cambios configurado.
Ten en cuenta que, si deseas que se muestren las columnas modificadas y no modificadas en cada
evento de cambio de datos, usa el tipo de captura de valor NEW_ROW. Para obtener más información, consulta Tipo de captura de valor.
Instala el JAR del conector de Kafka
Con Zookeeper, Kafka y Kafka Connect instalados, las tareas restantes para implementar un conector de Kafka son descargar el archivo del complemento del conector, extraer los archivos JAR en tu entorno de Kafka Connect y agregar el directorio con los archivos JAR a plugin.path de Kafka Connect.
Luego, debes reiniciar el proceso de Kafka Connect para que detecte los nuevos archivos JAR.
Si trabajas con contenedores inmutables, puedes extraer imágenes Imágenes de contenedor de Debezium para Zookeeper, Kafka y Kafka Connect. La imagen de Kafka Connect tiene preinstalado el conector de Spanner.
Para obtener más información sobre cómo instalar archivos JAR de conectores de Kafka basados en Debezium, consulta Cómo instalar Debezium.
Configura el conector de Kafka
El siguiente es un ejemplo de la configuración para un conector de Kafka
que se conecta a un flujo de cambios llamado changeStreamAll en el
La base de datos users en la instancia test-instance y el proyecto test-project.
"name": "spanner-connector", "config": { "connector.class": "io.debezium.connector.spanner.SpannerConnector", "gcp.spanner.project.id": "test-project", "gcp.spanner.instance.id": "test-instance", "gcp.spanner.database.id": "users", "gcp.spanner.change.stream": "changeStreamAll", "gcp.spanner.credentials.json": "{"client_id": user@example.com}", "gcp.spanner.database.role": "cdc-role", "tasks.max": "10" }
Esta configuración contiene lo siguiente:
El nombre del conector cuando se registró en un servicio de Kafka Connect.
Es el nombre de esta clase de conector de Spanner.
El ID del proyecto
El ID de la instancia de Spanner.
El ID de la base de datos de Spanner.
El nombre del flujo de cambios.
El objeto JSON de la clave de la cuenta de servicio.
El rol de base de datos de Spanner que se usará (opcional).
La cantidad máxima de tareas.
Para obtener una lista completa de las propiedades del conector, consulta Propiedades de configuración del conector de Kafka.
Agrega la configuración del conector a Kafka Connect
Sigue estos pasos para comenzar a ejecutar un conector de Spanner:
Crear una configuración para el conector de Spanner
Usa la API de REST de Kafka Connect para agregar esa configuración de conector a tu clúster de Kafka Connect.
Puedes enviar esta configuración con un comando POST a un servicio Kafka Connect en ejecución. De forma predeterminada, el servicio de Kafka Connect se ejecuta en el puerto 8083.
El servicio registra la configuración y comienza la tarea del conector que se conecta a la base de datos de Spanner y transmite registros de eventos de cambio a los temas de Kafka.
El siguiente es un ejemplo de comando POST:
POST /connectors HTTP/1.1
Host: http://localhost:8083
Accept: application/json
{
"name": "spanner-connector"
"config": {
"connector.class": "io.debezium.connector.spanner.SpannerConnector",
"gcp.spanner.project.id": "test-project",
"gcp.spanner.instance.id": "test-instance",
"gcp.spanner.database.id": "users",
"gcp.spanner.change.stream": "changeStreamAll",
"gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
"heartbeat.interval.ms": "100",
"tasks.max": "10"
}
}Ejemplo de respuesta correcta:
HTTP/1.1 201 Created
Content-Type: application/json
{
"name": "spanner-connector",
"config": {
"connector.class": "io.debezium.connector.spanner.SpannerConnector",
"gcp.spanner.project.id": "test-project",
"gcp.spanner.instance.id": "test-instance",
"gcp.spanner.database.id": "users",
"gcp.spanner.change.stream": "changeStreamAll",
"gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
"heartbeat.interval.ms": "100",
"tasks.max": "10"
},
"tasks": [
{ "connector": "spanner-connector", "task": 1 },
{ "connector": "spanner-connector", "task": 2 },
{ "connector": "spanner-connector", "task": 3 }
]
}Actualiza la configuración del conector de Kafka
Para actualizar la configuración del conector, envía un comando PUT al entorno
Servicio Kafka Connect con el mismo nombre de conector.
Supongamos que tenemos un conector en ejecución con la configuración de la sección anterior. El siguiente es un ejemplo de comando PUT:
PUT /connectors/spanner-connector/config HTTP/1.1
Host: http://localhost:8083
Accept: application/json
{
"connector.class": "io.debezium.connector.spanner.SpannerConnector",
"gcp.spanner.project.id": "test-project",
"gcp.spanner.instance.id": "test-instance",
"gcp.spanner.database.id": "users",
"gcp.spanner.change.stream": "changeStreamAll",
"gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
"heartbeat.interval.ms": "100",
"tasks.max": "10"
}Ejemplo de una respuesta correcta:
HTTP/1.1 200 OK
Content-Type: application/json
{
"connector.class": "io.debezium.connector.spanner.SpannerConnector",
"tasks.max": "10",
"gcp.spanner.project.id": "test-project",
"gcp.spanner.instance.id": "test-instance",
"gcp.spanner.database.id": "users",
"gcp.spanner.change.stream": "changeStreamAll",
"gcp.spanner.credentials.json": "{\"client_id\": \"XXXX\".... }",
"heartbeat.interval.ms": "100",
"tasks.max": "10"
}Detén el conector de Kafka
Para detener el conector, envía un comando DELETE al
Servicio Kafka Connect con el mismo nombre de conector.
Supongamos que tenemos un conector en ejecución con la configuración del
la sección anterior. El siguiente es un ejemplo de comando DELETE:
DELETE /connectors/spanner-connector HTTP/1.1 Host: http://localhost:8083
Ejemplo de una respuesta correcta:
HTTP/1.1 204 No Content
Supervisa el conector de Kafka
Además de las métricas estándar de Kafka Connect y Debezium, el conector de Kafka exporta sus propias métricas:
MilliSecondsLowWatermark: Es la marca de agua baja actual de la tarea del conector en milisegundos. El la marca de agua baja describe el tiempo T en el que se garantiza que el conector tenga transmitió todos los eventos con la marca de tiempo < MMilliSecondsLowWatermarkLag: Es la demora de la marca de agua baja con respecto a la hora actual, expresada en milisegundos. Transmitió todos los eventos con una marca de tiempo inferior a T.LatencyLowWatermark<Variant>MilliSeconds: Es la demora de la marca de agua baja con respecto a la hora actual en milisegundos. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Máximo.LatencySpanner<Variant>MilliSeconds: La latencia de lectura de la marca de tiempo de confirmación de Spanner al conector. Se proporcionan las variantes P50, P95, P99, Promedio, Mín. y Max.LatencyReadToEmit<Variant>MilliSeconds: Es la latencia de la emisión de la marca de tiempo de lectura de Spanner. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Máximo.LatencyCommitToEmit<Variant>tMilliSeconds: La latencia de Spanner-commit-timestamp-to-connector-emit. Se proporcionan las variantes P50, P95, P99, Average, Min y Max.LatencyCommitToPublish<Variant>MilliSeconds: Es la latencia de Spanner-commit-timestamp-to Kafka-publish-timestamp. Se proporcionan las variantes P50, P95, P99, Promedio, Mín. y Max.NumberOfChangeStreamPartitionsDetected: Es la cantidad total de particiones que detectó la tarea del conector actual.NumberOfChangeStreamQueriesIssued: La cantidad total de consultas de flujos de cambios emitidos por la tarea actual.NumberOfActiveChangeStreamQueries: Es la cantidad activa de flujos de cambios. las consultas detectadas por la tarea del conector actual.SpannerEventQueueCapacity: Es la capacidad total deStreamEventQueue, una cola que almacena elementos recibidos de las consultas de flujo de cambios.SpannerEventQueueCapacity: Es la capacidad restante deStreamEventQueue.TaskStateChangeEventQueueCapacity: La capacidad total deTaskStateChangeEventQueue, una cola que almacena eventos que suceden en el conector.RemainingTaskStateChangeEventQueueCapacity: La capacidad restante deTaskStateChangeEventQueue.NumberOfActiveChangeStreamQueries: Es la cantidad activa de consultas de transmisión de cambios que detecta la tarea del conector actual.
Propiedades de configuración del conector de Kafka
A continuación, se indican las propiedades de configuración obligatorias del conector:
name: Es el nombre único del conector. Si intentas registrarte de nuevo con el mismo nombre, fallará la ejecución. Todos los conectores de Kafka Connect requieren esta propiedad.connector.class: Es el nombre de la clase de Java del conector. Siempre usa un valor deio.debezium.connector.spanner.SpannerConnectorpara el conector de Kafka.tasks.max: La cantidad máxima de tareas que se deben crear para este conector.gcp.spanner.project.id: el ID del proyectogcp.spanner.instance.id: El ID de la instancia de Spannergcp.spanner.database.id: Es el ID de la base de datos de Spanner.gcp.spanner.change.stream: Es el nombre del flujo de cambios de Spanner.gcp.spanner.credentials.json: Es el objeto JSON de la clave de la cuenta de servicio.gcp.spanner.credentials.path: Es la ruta de acceso al archivo del objeto JSON de la clave de la cuenta de servicio. Es obligatorio si no se completa el campo anterior.gcp.spanner.database.role: El rol de base de datos de Spanner para usar. Esto solo es necesario cuando el flujo de cambios está protegido con un control de acceso detallado. El rol de la base de datos debe tener el privilegioSELECTen la transmisión de cambios y el privilegioEXECUTEen la función de lectura de la transmisión de cambios. Para obtener más información, consulta Control de acceso detallado para transmisiones continuas.
Las siguientes propiedades de configuración avanzada tienen valores predeterminados que funcionan en la mayoría de las situaciones y, por lo tanto, rara vez se deben especificar en la configuración del conector:
gcp.spanner.low-watermark.enabled: Indica si la marca de agua baja está habilitada para el conector. La configuración predeterminada es False.gcp.spanner.low-watermark.update-period.ms: Es el intervalo en el que se actualiza la marca de agua baja. El valor predeterminado es 1,000 ms.heartbeat.interval.ms: Es el intervalo de intervalo de Spanner. El valor predeterminado es 300,000 (cinco minutos).gcp.spanner.start.time: Es la hora de inicio del conector. El valor predeterminado es la hora actual.gcp.spanner.end.time: Es la hora de finalización del conector. La configuración predeterminada es infinito.tables.exclude.list: Son las tablas para las que se excluirán los eventos de cambio. La configuración predeterminada es vacía.tables.include.list: Son las tablas para las que se incluirán los eventos de cambio. Si no se propaga, se incluyen todas las tablas. La configuración predeterminada es vacía.gcp.spanner.stream.event.queue.capacity: Es la capacidad de la cola de eventos de Spanner. La configuración predeterminada es 10,000.connector.spanner.task.state.change.event.queue.capacity: Es la capacidad de la cola de eventos de cambio de estado de la tarea. La configuración predeterminada es 1,000.connector.spanner.max.missed.heartbeats: Es la cantidad máxima de señales de monitoreo de funcionamiento perdidas para una consulta de flujo de cambios antes de que se arroje una excepción. La configuración predeterminada es 10.scaler.monitor.enabled: Indica si el ajuste de escala automático de tareas está habilitado. La configuración predeterminada es "false".tasks.desired.partitions: Es la cantidad preferida de particiones de flujos de cambios por tarea. Este parámetro es necesario para el ajuste de escala automático de tareas. La configuración predeterminada es 2.tasks.min: Es la cantidad mínima de tareas. Este parámetro es necesario para el ajuste de escala automático de tareas. El valor predeterminado es 1.connector.spanner.sync.topic: Es el nombre del tema de sincronización, un tema de conector interno que se usa para almacenar la comunicación entre tareas. El valor predeterminado es_sync_topic_spanner_connector_connectornamesi el usuario no proporcionó un nombre.connector.spanner.sync.poll.duration: la duración de la encuesta del tema de sincronización. El valor predeterminado es de 500 ms.connector.spanner.sync.request.timeout.ms: Es el tiempo de espera para las solicitudes al tema de sincronización. El valor predeterminado es 5,000 ms.connector.spanner.sync.delivery.timeout.ms: Es el tiempo de espera para publicar en el tema de sincronización. La configuración predeterminada es 15,000 ms.connector.spanner.sync.commit.offsets.interval.ms: El intervalo en el que se confirman los desplazamientos para el tema de sincronización. La configuración predeterminada es 60,000 ms.connector.spanner.sync.publisher.wait.timeout: Es el intervalo en el que se publican los mensajes en el tema de sincronización. La configuración predeterminada es de 5 ms.connector.spanner.rebalancing.topic: Es el nombre del tema de reabastecimiento. El tema de reabastecimiento es un tema de conector interno que se usa para determinar el estado activo de la tarea. El valor predeterminado es_rebalancing_topic_spanner_connector_connectornamesi el usuario no proporcionó un nombre.connector.spanner.rebalancing.poll.duration: Es la duración de la encuesta para el tema de reabastecimiento. La configuración predeterminada es 5,000 ms.connector.spanner.rebalancing.commit.offsets.timeout: Es el tiempo de espera para confirmar compensaciones para el tema de rebalanceo. La configuración predeterminada es 5,000 ms.connector.spanner.rebalancing.commit.offsets.interval.ms: El intervalo en el que se confirman los desplazamientos para el tema de sincronización. El valor predeterminado es 60,000 ms.connector.spanner.rebalancing.task.waiting.timeout: Es la duración del tiempo que una tarea espera antes de procesar un evento de reabastecimiento. El valor predeterminado es 1,000 ms.
Para obtener una lista aún más detallada de las propiedades del conector configurables, consulta el repositorio de GitHub.
Limitaciones
El conector no admite la transmisión de eventos de instantáneas.
Si la marca de agua está habilitada en el conector, no puede configurar el tema Debezium transformaciones de enrutamiento.