Esta página explica como usar o conetor do Kafka para consumir e encaminhar dados de streams de alterações do Spanner.
Conceitos principais
A secção seguinte descreve os conceitos principais do conetor do Kafka.
Debezium
O Debezium é um projeto de código aberto que oferece uma plataforma de streaming de dados de baixa latência para a captura de dados de alterações.
Conetor do Kafka
O conetor Kafka oferece uma abstração sobre a API Spanner para publicar streams de alterações do Spanner no Kafka. Com este conector, não tem de gerir o ciclo de vida da partição de streams de alterações, o que é necessário quando usa a API Spanner diretamente.
O conector do Kafka produz um evento de alteração para cada registo de modificação de dados e envia registos de eventos de alteração a jusante para um tópico do Kafka separado para cada tabela monitorizada pela stream de alterações. Um registo de alteração de dados mod representa uma única modificação (inserção, atualização ou eliminação) que foi captada. Um único registo de alteração de dados pode conter mais do que uma modificação.
Resultado do conetor do Kafka
O conector Kafka encaminha os registos de streams de alterações diretamente
para um tópico Kafka separado. O nome do tópico de saída deve ser connector_name.table_name.
Se o tópico não existir, o conector do Kafka cria automaticamente um tópico com esse nome.
Também pode configurar transformações de encaminhamento de tópicos para reencaminhar registos para tópicos que especificar. Se quiser usar o encaminhamento de tópicos, desative a funcionalidade de marca de água baixa.
Ordenação de registos
Os registos são ordenados pela data/hora de confirmação por chave primária nos tópicos do Kafka. Os registos pertencentes a chaves primárias diferentes não têm garantias de ordenação. Os registos com a mesma chave primária são armazenados na mesma partição do tópico do Kafka. Se quiser processar transações completas, também pode usar os campos data change record's
server_transaction_id e number_of_records_in_transaction para
criar uma transação do Spanner.
Altere eventos
O conetor do Kafka gera um evento de alteração de dados para cada operação INSERT, UPDATE e DELETE. Cada evento contém uma chave e valores para a linha alterada.
Pode usar conversores do Kafka Connect para produzir eventos de alteração de dados nos formatos Protobuf, AVRO, JSON ou JSON Schemaless. Se usar um conversor do Kafka Connect que produza esquemas, o evento contém esquemas separados para a chave e os valores. Caso contrário, o evento contém apenas a chave e os valores.
O esquema da chave nunca muda. O esquema dos valores é uma combinação de todas as colunas que a stream de alterações acompanhou desde a hora de início do conector.
Se configurar o conetor para produzir eventos JSON, o evento de alteração de saída contém cinco campos:
O primeiro campo
schemaespecifica um esquema do Kafka Connect que descreve o esquema de chaves do Spanner.O primeiro campo
payloadtem a estrutura descrita pelo camposchemaanterior e contém a chave da linha que foi alterada.O segundo campo
schemaespecifica o esquema do Kafka Connect que descreve o esquema da linha alterada.O segundo campo
payloadtem a estrutura descrita pelo camposchemaanterior e contém os dados reais da linha que foi alterada.O campo
sourceé um campo obrigatório que descreve os metadados de origem do evento.
Segue-se um exemplo de um evento de alteração de dados:
{
// 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 água baixa
A marca d'água baixa descreve a hora T em que o conector do Kafka tem a garantia de ter transmitido e publicado num tópico do Kafka todos os eventos com a data/hora < T.
Pode ativar a marca de água baixa no conetor do Kafka através do parâmetro gcp.spanner.low-watermark.enabled. Este parâmetro está desativado por predefinição. Se a marca d'água baixa estiver ativada, o campo low_watermark no registo de alterações dos dados do fluxo de alterações é preenchido com a data/hora atual da marca d'água baixa do conetor do Kafka.
Se não estiverem a ser produzidos registos, o conetor do Kafka envia "heartbeats" de marca de água periódicos para os tópicos de saída do Kafka detetados pelo conetor.
Estes sinais de áudio de marca de água são registos vazios, exceto para o campo low_watermark. Em seguida, pode usar a marca de água inferior para fazer agregações baseadas no tempo.
Por exemplo, pode usar a marca d'água baixa para ordenar eventos por data/hora de confirmação nas chaves primárias.
Tópicos de metadados
O conetor do Kafka, bem como a framework Kafka Connect, cria vários tópicos de metadados para armazenar informações relacionadas com o conetor. Não é aconselhável modificar a configuração nem o conteúdo destes tópicos de metadados.
Seguem-se os tópicos de metadados:
_consumer_offsets: um tópico criado automaticamente pelo Kafka. Armazena as compensações do consumidor para os consumidores criados no conector do Kafka._kafka-connect-offsets: um tópico criado automaticamente pelo Kafka Connect. Armazena os desvios do conetor._sync_topic_spanner_connector_connectorname: um tópico criado automaticamente pelo conector. Armazena metadados relativos a partições da stream de alterações._rebalancing_topic_spanner_connector_connectorname: um tópico criado automaticamente pelo conector. Usado para determinar a atividade da tarefa do conetor._debezium-heartbeat.connectorname: um tópico usado para processar os sinais de pulsação da stream de alterações do Spanner.
Tempo de execução do conetor do Kafka
A descrição seguinte descreve o tempo de execução do conector do Kafka.
Escalabilidade
O conetor do Kafka é horizontalmente escalável e é executado em uma ou mais tarefas distribuídas por vários trabalhadores do Kafka Connect.
Garantias de entrega de mensagens
O conetor do Kafka suporta a garantia de entrega pelo menos uma vez.
Tolerância a falhas
O conetor do Kafka é tolerante a falhas. À medida que o conetor do Kafka lê as alterações e produz eventos, regista a data/hora da última confirmação processada para cada partição do fluxo de alterações. Se o conetor do Kafka parar por qualquer motivo (incluindo falhas de comunicação, problemas de rede ou falhas de software), após o reinício, o conetor do Kafka continua a transmitir registos a partir do ponto em que parou.
O conetor do Kafka lê o esquema de informações na data/hora de início do conetor do Kafka para obter informações do esquema. Por predefinição, o Spanner não consegue ler o esquema de informações em datas/horas de leitura anteriores ao período de retenção de versões, que é de uma hora por predefinição. Se quiser iniciar o conector antes de uma hora, tem de aumentar o período de retenção de versões da base de dados.
Configure o conetor do Kafka
Crie uma stream de alterações
Para ver detalhes sobre como criar uma stream de alterações, consulte o artigo Crie uma stream de alterações. Para continuar com os passos seguintes, é necessária uma instância do Spanner com uma stream de alterações configurada.
Tenha em atenção que, se quiser que as colunas alteradas e não alteradas sejam devolvidas em cada evento de alteração de dados, use o tipo de captura de valor NEW_ROW. Para mais informações, consulte o tipo de captura de valor.
Instale o JAR do conetor do Kafka
Com o Zookeeper, o Kafka e o Kafka Connect instalados, as tarefas restantes para implementar um conetor do Kafka são transferir
o arquivo de plug-ins do conetor, extrair os ficheiros JAR para o seu ambiente do Kafka Connect e adicionar o
diretório com os ficheiros JAR ao plugin.path do Kafka Connect.
Em seguida, tem de reiniciar o processo do Kafka Connect para selecionar os novos ficheiros JAR.
Se estiver a trabalhar com contentores imutáveis, pode extrair imagens dos contentores de imagens do Debezium para o Zookeeper, o Kafka e o Kafka Connect. A imagem do Kafka Connect tem o conetor do Spanner pré-instalado.
Para mais informações sobre como instalar JARs do conetor Kafka baseados no Debezium, consulte o artigo Instalar o Debezium.
Configure o conetor do Kafka
Segue-se um exemplo da configuração de um conetor do Kafka que se liga a uma stream de alterações denominada changeStreamAll na base de dados users na instância test-instance e no projeto 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 configuração contém o seguinte:
O nome do conetor quando registado num serviço Kafka Connect.
O nome desta classe de conetor do Spanner.
O ID do projeto.
O ID da instância do Spanner.
O ID da base de dados do Spanner.
O nome da stream de alterações.
O objeto JSON para a chave da conta de serviço.
(Opcional) A função da base de dados do Spanner a usar.
O número máximo de tarefas.
Para ver uma lista completa das propriedades do conetor, consulte o artigo Propriedades de configuração do conetor do Kafka.
Adicione a configuração do conetor ao Kafka Connect
Para começar a executar um conetor do Spanner:
Crie uma configuração para o conetor do Spanner.
Use a API REST do Kafka Connect para adicionar essa configuração do conetor ao cluster do Kafka Connect.
Pode enviar esta configuração com um comando POST para um serviço Kafka Connect em execução. Por predefinição, o serviço Kafka Connect é executado na porta 8083.
O serviço regista a configuração e inicia a tarefa do conetor que se liga à base de dados do Spanner e transmite registos de eventos de alteração para tópicos do Kafka.
Segue-se um exemplo de um 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"
}
}Exemplo de resposta com êxito:
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 }
]
}Atualize a configuração do conetor do Kafka
Para atualizar a configuração do conetor, envie um comando PUT para o serviço Kafka Connect em execução com o mesmo nome do conetor.
Parta do princípio de que temos um conector em execução com a configuração da secção anterior. Segue-se um exemplo de um 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"
}Exemplo de resposta com êxito:
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"
}Pare o conetor do Kafka
Para parar o conector, envie um comando DELETE para o serviço Kafka Connect em execução com o mesmo nome do conector.
Parta do princípio de que temos um conector em execução com a configuração da secção anterior. Segue-se um exemplo de um comando DELETE:
DELETE /connectors/spanner-connector HTTP/1.1 Host: http://localhost:8083
Exemplo de resposta com êxito:
HTTP/1.1 204 No Content
Monitorize o conetor do Kafka
Além das métricas padrão do Kafka Connect e do Debezium, o conetor do Kafka exporta as suas próprias métricas:
MilliSecondsLowWatermark: a marca de água baixa atual da tarefa do conector em milissegundos. A marca d'água inferior descreve a hora T em que o conector tem a garantia de ter transmitido todos os eventos com a data/hora < TMilliSecondsLowWatermarkLag: O atraso da marca de água baixa em relação à hora atual, em milissegundos. streamed out all events with timestamp < TLatencyLowWatermark<Variant>MilliSeconds: o atraso da marca de água baixa em relação à hora atual, em milissegundos. São fornecidas as variantes P50, P95, P99, média, mínima e máxima.LatencySpanner<Variant>MilliSeconds: a latência de Spanner-commit-timestamp-to-connector-read. São fornecidas as variantes P50, P95, P99, média, mínima e máxima.LatencyReadToEmit<Variant>MilliSeconds: a latência de leitura da data/hora do Spanner para a emissão do conector. São fornecidas as variantes P50, P95, P99, média, mínima e máxima.LatencyCommitToEmit<Variant>tMilliSeconds: a latência de Spanner-commit-timestamp-to-connector-emit. São fornecidas as variantes P50, P95, P99, média, mínima e máxima.LatencyCommitToPublish<Variant>MilliSeconds: a latência da indicação de tempo de confirmação do Spanner para a indicação de tempo de publicação do Kafka. São fornecidas as variantes P50, P95, P99, média, mínima e máxima.NumberOfChangeStreamPartitionsDetected: o número total de partições detetadas pela tarefa do conector atual.NumberOfChangeStreamQueriesIssued: o número total de consultas de streams de alterações emitidas pela tarefa atual.NumberOfActiveChangeStreamQueries: o número ativo de consultas de fluxo de alterações detetadas pela tarefa do conector atual.SpannerEventQueueCapacity: a capacidade total deStreamEventQueue, uma fila que armazena elementos recebidos de consultas de streams de alterações.SpannerEventQueueCapacity: a capacidade restante deStreamEventQueue.TaskStateChangeEventQueueCapacity: a capacidade total deTaskStateChangeEventQueue, uma fila que armazena eventos que ocorrem no conector.RemainingTaskStateChangeEventQueueCapacity: a capacidade restante deTaskStateChangeEventQueue.NumberOfActiveChangeStreamQueries: o número ativo de consultas de fluxo de alterações detetadas pela tarefa do conector atual.
Propriedades de configuração do conetor do Kafka
Seguem-se as propriedades de configuração necessárias para o conetor:
name: nome exclusivo do conector. A tentativa de registo novamente com o mesmo nome provoca uma falha. Esta propriedade é necessária para todos os conetores do Kafka Connect.connector.class: o nome da classe Java para o conetor. Use sempre um valor deio.debezium.connector.spanner.SpannerConnectorpara o conetor do Kafka.tasks.max: o número máximo de tarefas que devem ser criadas para este conector.gcp.spanner.project.id: o ID do projetogcp.spanner.instance.id: o ID da instância do Spannergcp.spanner.database.id: o ID da base de dados do Spannergcp.spanner.change.stream: o nome do fluxo de alterações do Spannergcp.spanner.credentials.json: o objeto JSON da chave da conta de serviço.gcp.spanner.credentials.path: o caminho do ficheiro para o objeto JSON da chave da conta de serviço. Obrigatório se o campo acima não for fornecido.gcp.spanner.database.role: a função da base de dados do Spanner a usar. Isto só é necessário quando a stream de alterações está protegida com o controlo de acesso detalhado. A função de base de dados tem de ter o privilégioSELECTna stream de alterações e o privilégioEXECUTEna função de leitura da stream de alterações. Para mais informações, consulte o artigo Controlo de acesso detalhado para streams de alterações.
As seguintes propriedades de configuração avançadas têm predefinições que funcionam na maioria das situações e, por isso, raramente precisam de ser especificadas na configuração do conetor:
gcp.spanner.low-watermark.enabled: indica se a marca de água baixa está ativada para o conector. O valor predefinido é False.gcp.spanner.low-watermark.update-period.ms: o intervalo no qual a marca de água baixa é atualizada. A predefinição é 1000 ms.heartbeat.interval.ms: o intervalo de batimentos cardíacos do Spanner. A predefinição é 300 000 (cinco minutos).gcp.spanner.start.time: a hora de início do conector. A predefinição é a hora atual.gcp.spanner.end.time: a hora de fim do conetor. A predefinição é infinito.tables.exclude.list: as tabelas para as quais excluir eventos de alteração. A predefinição é vazio.tables.include.list: as tabelas para as quais incluir eventos de alteração. Se não for preenchido, todas as tabelas são incluídas. A predefinição é vazio.gcp.spanner.stream.event.queue.capacity: a capacidade da fila de eventos do Spanner. A predefinição é 10 000.connector.spanner.task.state.change.event.queue.capacity: a capacidade da fila de eventos de alteração do estado da tarefa. A predefinição é 1000.connector.spanner.max.missed.heartbeats: o número máximo de pulsações perdidas para uma consulta de stream de alterações antes de ser gerada uma exceção. A predefinição é 10.scaler.monitor.enabled: indica se o dimensionamento automático de tarefas está ativado. A predefinição é false.tasks.desired.partitions: o número preferencial de partições de streams de alterações por tarefa. Este parâmetro é necessário para o dimensionamento automático de tarefas. A predefinição é 2.tasks.min: o número mínimo de tarefas. Este parâmetro é necessário para o dimensionamento automático de tarefas. A predefinição é 1.connector.spanner.sync.topic: o nome do tópico de sincronização, um tópico de conector interno usado para armazenar a comunicação entre tarefas. A predefinição é_sync_topic_spanner_connector_connectornamese o utilizador não tiver indicado um nome.connector.spanner.sync.poll.duration: a duração da sondagem para o tópico de sincronização. A predefinição é 500 ms.connector.spanner.sync.request.timeout.ms: o limite de tempo para pedidos ao tópico de sincronização. A predefinição é 5000 ms.connector.spanner.sync.delivery.timeout.ms: o tempo limite para publicação no tópico de sincronização. A predefinição é 15 000 ms.connector.spanner.sync.commit.offsets.interval.ms: o intervalo no qual os desvios são confirmados para o tópico de sincronização. A predefinição é 60 000 ms.connector.spanner.sync.publisher.wait.timeout: o intervalo no qual as mensagens são publicadas no tópico de sincronização. A predefinição é 5 ms.connector.spanner.rebalancing.topic: o nome do tópico de reequilíbrio. O tópico de reequilíbrio é um tópico de conetor interno usado para determinar a atividade das tarefas. A predefinição é_rebalancing_topic_spanner_connector_connectornamese o utilizador não tiver indicado um nome.connector.spanner.rebalancing.poll.duration: a duração da sondagem para o tópico de reequilíbrio. A predefinição é 5000 ms.connector.spanner.rebalancing.commit.offsets.timeout: o limite de tempo para confirmar os desvios do tópico de reequilíbrio. A predefinição é 5000 ms.connector.spanner.rebalancing.commit.offsets.interval.ms: o intervalo no qual os desvios são confirmados para o tópico de sincronização. A predefinição é 60 000 ms.connector.spanner.rebalancing.task.waiting.timeout: a duração do tempo que uma tarefa aguarda antes de processar um evento de reequilíbrio. A predefinição é 1000 ms.
Para uma lista ainda mais detalhada das propriedades configuráveis do conetor, consulte o repositório do GitHub.
Limitações
O conetor não suporta o streaming de eventos de captura de ecrã.
Se a marca de água estiver ativada no conector, não pode configurar transformações de encaminhamento de tópicos do Debezium.