En esta página, se explica cómo usar el conector de Kafka para consumir y reenviar datos de flujos de cambios de Spanner.
Conceptos básicos
A continuación, se describen los conceptos básicos 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 de Kafka proporciona una abstracción sobre la API de Spanner para publicar flujos de cambios de Spanner en Kafka. Con este conector, no tienes que administrar el ciclo de vida de la partición de flujos de cambios, lo cual 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 cambios de datos y envía registros de eventos de cambios de forma descendente a un tema de Kafka separado para cada tabla de seguimiento de cambios de la que se realiza el seguimiento. Una modificación 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 cambios de datos puede contener más de una modificación.
Salida del conector de Kafka
El conector de Kafka reenvía los registros de flujos de cambios directamente a un tema de Kafka separado. El nombre del tema de salida debe ser connector_name.table_name. Si el tema no existe, el conector de Kafka crea de forma automática un tema con ese nombre.
También puedes configurar transformaciones de enrutamiento de temas para redireccionar los registros a temas que especifiques. Si deseas utilizar el enrutamiento de temas, inhabilita la funcionalidad de marca de agua baja.
Orden de los registros
Los registros se ordenan según la 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 orden. Los registros con la misma clave primaria se almacenan en la misma partición del tema de Kafka. Si deseas procesar transacciones completas, también puedes usar los campos server_transaction_id y number_of_records_in_transaction del registro de cambios de datos para ensamblar una transacción de Spanner.
Eventos de cambio
El conector de Kafka genera un evento de cambio de datos para cada operación INSERT, UPDATE y DELETE. Cada evento contiene una clave y valores para la fila modificada.
Puedes usar los convertidores 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 para los valores es una combinación de todas las columnas a las que se realizó un seguimiento del flujo de cambios 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 modificó.El segundo campo
schemaespecifica el esquema Kafka Connect, que describe el esquema para 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 obligatorio y 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 la marca de tiempo < T.
Puedes habilitar la marca de agua baja en el conector de Kafka mediante 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 los datos del flujo de cambios se propaga con la marca de tiempo de marca de agua baja actual del conector de Kafka.
Si no se producen registros, el conector de Kafka envía de forma periódica “latidos” de la marca de agua a los temas de salida de Kafka que detecta el conector.
Estas señales de marca de agua son registros vacíos, excepto por el campo 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 eventos por marca de tiempo de confirmación en todas las claves primarias.
Temas de metadatos
El conector de Kafka, al igual que el framework de Kafka Connect, crean varios temas de metadatos para almacenar información relacionada con el conector. No se recomienda modificar la configuración ni el contenido de estos temas de metadatos.
Los siguientes son los temas de metadatos:
_consumer_offsets: Un tema que Kafka crea automáticamente. Almacena compensaciones para los consumidores creadas en el conector de Kafka._kafka-connect-offsets: Un tema creado automáticamente por Kafka Connect. Almacena los desplazamientos del conector._sync_topic_spanner_connector_connectorname: El conector crea un tema automáticamente. Almacena metadatos relacionados con particiones de flujos de cambios._rebalancing_topic_spanner_connector_connectorname: El conector crea un tema automáticamente. Se usa para determinar la actividad de la tarea del conector._debezium-heartbeat.connectorname: Es un tema que se usa para procesar las señales de monitoreo de funcionamiento 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 los errores. 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 (incluidas las fallas de comunicación, los problemas de red o las fallas de software), el conector de Kafka continúa transmitiendo los registros desde la última vez que se detuvo.
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 las 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 deseas iniciar el conector desde hace más de una hora, debes aumentar el período de retención de la versión de la base de datos.
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 próximos 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 sin cambios en cada evento de cambio de datos, usa el tipo de captura de valor NEW_ROW. Para obtener más información, consulta el tipo de captura de valores.
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 de complementos del conector, extraer los archivos JAR en tu entorno de Kafka Connect y agregar el directorio con los archivos JAR al plugin.path de Kafka Connect.
Luego, debes reiniciar el proceso de Kafka Connect para recoger los archivos JAR nuevos.
Si trabajas con contenedores inmutables, puedes extraer imágenes de 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 los archivos JAR del conector de Kafka basados en Debezium, consulta Instala Debezium.
Configura el conector de Kafka
El siguiente es un ejemplo de la configuración de un conector de Kafka que se conecta a un flujo de cambios llamado changeStreamAll en 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.
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 para 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 de 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, luego, inicia la tarea del conector que se conecta a la base de datos de Spanner y transmite los 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 servicio de Kafka en ejecución 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 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 de Kafka en ejecución 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 DELETE:
DELETE /connectors/spanner-connector HTTP/1.1 Host: http://localhost:8083
Ejemplo de 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. La marca de agua baja describe la hora T en la que se garantiza que el conector transmitió todos los eventos con la marca de tiempo < T.MilliSecondsLowWatermarkLag: El retraso de la marca de agua baja con respecto a la hora actual en milisegundos. Se transmitieron todos los eventos con una marca de tiempo < TLatencyLowWatermark<Variant>MilliSeconds: El retraso de la marca de agua baja con respecto a la hora actual en milisegundos. Se proporcionan las variantes P50, P95, P99, Average, Min y Max.LatencySpanner<Variant>MilliSeconds: Es la latencia de lectura de Spanner-commit-timestamp-to-connector-read. 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, Average, Min y Max.LatencyCommitToEmit<Variant>tMilliSeconds: Es 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: La cantidad total de particiones detectadas por la tarea del conector actual.NumberOfChangeStreamQueriesIssued: La cantidad total de consultas del flujo de cambios que emite la tarea actual.NumberOfActiveChangeStreamQueries: La cantidad activa de consultas de flujos de cambios detectadas por la tarea del conector actual.SpannerEventQueueCapacity: La capacidad total deStreamEventQueue, una cola que almacena elementos recibidos de consultas de flujos de cambios.SpannerEventQueueCapacity: 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: La cantidad activa de consultas de flujos de cambios detectadas por 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: Nombre único para el 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 para el conector. Usa siempre 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 objeto JSON de la clave de la cuenta de servicio. Es obligatorio si no se completa el campo anterior.gcp.spanner.database.role: Es la función de base de datos de Spanner que se debe usar. Esto es necesario solo cuando el flujo de cambios está protegido con un control de acceso detallado. La función de la base de datos debe tener el privilegioSELECTen el flujo de cambios y el privilegioEXECUTEen la función de lectura del flujo de cambios. Si deseas obtener más información, consulta Control de acceso detallado para flujos de cambios.
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 deben especificarse 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: El intervalo de la señal de monitoreo de funcionamiento 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: La hora de finalización del conector. La configuración predeterminada es infinito.tables.exclude.list: Son las tablas para las que se excluyen 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: 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: 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. La configuración predeterminada es 500 ms.connector.spanner.sync.request.timeout.ms: Es el tiempo de espera para las solicitudes al tema de sincronización. La configuración predeterminada 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 rebalanceo. El tema de rebalanceo es un tema de conector interno que se usa para determinar la actividad 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 del tema de rebalanceo. 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. La configuración predeterminada es 60,000 ms.connector.spanner.rebalancing.task.waiting.timeout: El tiempo que espera una tarea antes de procesar un evento de rebalanceo. La configuración predeterminada es 1,000 ms.
Para obtener una lista aún más detallada de las propiedades de 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 puedes configurar las transformaciones de enrutamiento de tema de Debezium.