En esta página, se explica cómo usar el conector de Kafka para consumir y reenviar datos de flujos de cambio 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 los 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 cambio de manera descendente a un tema de Kafka independiente para cada tabla de seguimiento de transmisión de cambios. Una modificación de registro de cambios de datos representa una sola modificación (insertar, actualizar o borrar) que se capturó. Un solo registro de cambio 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 independiente. El nombre del tema de salida debe ser connector_name.table_name. Si no existe, el conector de Kafka crea un tema automáticamente con ese nombre.
También puedes configurar transformaciones de enrutamiento de temas para volver a enrutar los registros hacia los temas que especifiques. Si deseas 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 orden. Los registros con la misma clave primaria se almacenan en la misma partición de 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.
Cambiar eventos
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 convertidores de Kafka Connect para producir eventos de cambio de datos en formatos Protobuf, AVRO, JSON o JSON Schemaless. Si usas un convertidor de Kafka Connect que produce esquemas, el evento contendrá 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 que el flujo de cambios realizó desde la hora de inicio del conector.
Si configuras el conector para que produzca eventos JSON, el evento de cambio de salida contendrá cinco campos:
El primer campo
schemaespecifica un esquema de Kafka Connect que describe el esquema de clave 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 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 obligatorio y describe los metadatos de origen del evento.
A continuación, se muestra 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 el tiempo T en el que se garantiza que el conector de Kafka transmitirá y publicó todos los eventos con una marca de tiempo < T en un tema de Kafka.
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 los datos del flujo de cambios se propaga con la marca de tiempo actual de la marca de agua baja del conector de Kafka.
Si no se producen registros, el conector de Kafka envía señales de monitoreo de funcionamiento periódicas a los temas de salida de Kafka que detecta el conector.
Estas señales de monitoreo de funcionamiento de marcas de agua son registros que están vacíos, excepto en 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 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 temas de metadatos son los siguientes:
_consumer_offsets: Es un tema que Kafka creó automáticamente. Almacena las compensaciones de consumidores para los consumidores creados 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: Es un tema que el conector crea automáticamente. Almacena metadatos sobre particiones de flujos de cambios._rebalancing_topic_spanner_connector_connectorname: Es un tema que el conector crea automáticamente. Se usa para determinar la disponibilidad de la tarea del conector._debezium-heartbeat.connectorname: Es un tema que se usa para procesar las señales de monitoreo de funcionamiento de la transmisión 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 errores. A medida que el conector de Kafka lee los cambios y produce eventos, registra la última marca de tiempo de confirmación que se procesa para cada partición de flujo de cambios. Si el conector de Kafka se detiene por algún motivo (incluso fallas de comunicación, problemas de red o fallas de software), después de reiniciarlo, el conector de Kafka continúa transmitiendo registros desde donde se interrumpió por última vez.
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 antes del período de retención de la versión, que es de una hora de forma predeterminada. Si deseas iniciar el conector antes 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 información 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 las 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 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 de complementos del conector, extraer los archivos JAR al entorno de Kafka Connect y agregar el directorio con los archivos JAR al Kafka Connect plugin.path.
A continuación, debes reiniciar el proceso de Kafka Connect para recoger los nuevos archivos JAR.
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 JAR del conector de Kafka basado 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:
Es el nombre del conector cuando se registra en un servicio de Kafka Connect.
El nombre de esta clase de conector de Spanner.
El ID del proyecto
El ID de instancia de Spanner.
El ID de la base de datos de Spanner.
Es 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 las propiedades de configuración del conector de Kafka.
Agrega la configuración del conector a Kafka Connect
Para comenzar a ejecutar un conector de Spanner, haz lo siguiente:
Crea 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 registros de eventos de cambio a temas de Kafka.
El siguiente es un ejemplo del 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 Connect 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 del 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 Connect 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 del 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 el tiempo T en el que se garantiza que el conector transmitió todos los eventos con una marca de tiempo < T.MilliSecondsLowWatermarkLag: El retraso de la marca de agua baja detrás de la hora actual en milisegundos. Transmitió todos los eventos con una marca de tiempo < T.LatencyLowWatermark<Variant>MilliSeconds: El retraso de la marca de agua baja en milisegundos detrás de la hora actual. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Max.LatencySpanner<Variant>MilliSeconds: La latencia de Spanner-commit-timestamp-to-connector-read. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Máximo.LatencyReadToEmit<Variant>MilliSeconds: La latencia de Spanner-read-timestamp-to-connector-emit. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Max.LatencyCommitToEmit<Variant>tMilliSeconds: Es la latencia de Spanner-commit-timestamp-to-connector-emit. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Max.LatencyCommitToPublish<Variant>MilliSeconds: La latencia de Spanner-commit-timestamp-to Kafka-publish-timestamp. Se proporcionan las variantes P50, P95, P99, Promedio, Mínimo y Máximo.NumberOfChangeStreamPartitionsDetected: La cantidad total de particiones que detecta la tarea del conector actual.NumberOfChangeStreamQueriesIssued: La cantidad total de consultas de flujos de cambios que emite la tarea actual.NumberOfActiveChangeStreamQueries: Es 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 restanteStreamEventQueue.TaskStateChangeEventQueueCapacity: La capacidad total deTaskStateChangeEventQueue, una cola que almacena eventos que suceden en el conector.RemainingTaskStateChangeEventQueueCapacity: La capacidad restanteTaskStateChangeEventQueue.NumberOfActiveChangeStreamQueries: Es la cantidad activa de consultas de flujos de cambios detectadas por la tarea del conector actual.
Propiedades de configuración del conector de Kafka
Las siguientes son propiedades de configuración obligatorias para el conector:
name: El nombre único del conector. Si intentas volver a registrarte con el mismo nombre, se producirá un error. Esta propiedad es obligatoria para todos los conectores de Kafka Connect.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: El ID de la base de datos de Spannergcp.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 del archivo al objeto JSON de la clave de la cuenta de servicio. Obligatorio si no se proporciona el campo anterior.gcp.spanner.database.role: Es el rol de base de datos de Spanner que se usará. Esto es necesario solo cuando el flujo de cambios está protegido con control de acceso detallado. La función de 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 los flujos de cambios.
Las siguientes propiedades de configuración avanzada tienen valores predeterminados que funcionan en la mayoría de los casos y, por lo tanto, rara vez es necesario especificarlos en la configuración del conector:
gcp.spanner.low-watermark.enabled: Indica si la marca de agua baja está habilitada para el conector. El valor predeterminado es falso.gcp.spanner.low-watermark.update-period.ms: 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 señal de monitoreo de funcionamiento de Spanner. El valor predeterminado es 300000 (cinco minutos).gcp.spanner.start.time: Es la hora de inicio del conector. La configuración predeterminada es la hora actual.gcp.spanner.end.time: Es la hora de finalización del conector. La configuración predeterminada es infinita.tables.exclude.list: Son las tablas en las que se deben excluir los eventos de cambio. La configuración predeterminada es vacía.tables.include.list: Son las tablas para las que se incluirán 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: La capacidad de cola de eventos de Spanner. La configuración predeterminada es 10,000.connector.spanner.task.state.change.event.queue.capacity: El estado de la tarea cambia la capacidad de cola de eventos. 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 de 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 la tarea 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: Es la duración de la encuesta del tema de sincronización. La configuración predeterminada es 500 ms.connector.spanner.sync.request.timeout.ms: El tiempo de espera de las solicitudes al tema de sincronización. La configuración predeterminada es 5,000 ms.connector.spanner.sync.delivery.timeout.ms: 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 del 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 las tareas. El valor predeterminado es_rebalancing_topic_spanner_connector_connectornamesi el usuario no proporcionó un nombre.connector.spanner.rebalancing.poll.duration: La duración de la encuesta del tema de rebalanceo La configuración predeterminada es 5,000 ms.connector.spanner.rebalancing.commit.offsets.timeout: 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 del 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 del conector configurable, 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 temas de Debezium.