Esta página 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 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 cambios de datos y envía registros de eventos de cambio descendentes en un tema de Kafka separado para cada tabla de seguimiento de cambios. 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
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 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 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,
también 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 los convertidores de Kafka Connect para producir eventos de cambio de datos en
Formatos Protobuf, AVRO, JSON o JSON Schemaless. Si utilizas 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 un una amalgamación de todas las columnas a las que se hizo un seguimiento del flujo de cambios desde el tiempo 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 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 se habilita la marca de agua baja, el campo low_watermark de los datos del flujo de cambios
registro de cambios se completa con la marca de agua baja actual del conector de Kafka
y marca de tiempo.
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 confirmación
marca de tiempo entre 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 o 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: 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 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, produce eventos, registra la última marca de tiempo de confirmación procesada para cada cambio partición de transmisión. 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 al inicio del conector de Kafka. marca de tiempo para recuperar la información del esquema. De forma predeterminada, Spanner no puede lee el esquema de información en las marcas de tiempo de lectura antes de período de retención de la versión, que se establece en una hora de forma predeterminada. 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 próximos pasos, se requiere una instancia de Spanner con un flujo de cambios configurado.
Ten en cuenta que si deseas que se devuelvan 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 del complemento del conector, extrae los archivos JAR a tu entorno de Kafka Connect y agrega el
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 Imágenes de contenedor de Debezium para Zookeeper, Kafka y Kafka Connect. La imagen de Kafka Connect tiene el elemento El conector de Spanner está preinstalado.
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 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.
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 eso. configuración del conector al clúster de Kafka Connect.
Puedes enviar esta configuración con un comando POST a una conexión de Kafka en ejecución
servicio. 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
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 del
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 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 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 < TMilliSecondsLowWatermarkLag: El retraso de la marca de agua baja con respecto al tiempo actual en milisegundos. transmitió todos los eventos con la 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: Es la cantidad total de particiones. detectado por 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: 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: Es la cantidad activa de flujos de cambios. las consultas 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: El rol de base de datos de Spanner para usar. Esto es necesario solo 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 y el privilegioEXECUTEen el flujo 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 y, por lo tanto, rara vez debe 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 puede configurar el tema Debezium transformaciones de enrutamiento.