Conectar ao Lakebase

info

Visualização

Esse recurso está em Prévia Pública.

Use a transmissão estructurada para gravar no Lakebase com lotes integrada, novas tentativas automáticas e autenticação gerenciada pelo workspace.

Quando usar o destino do Lakebase

Use o coletor Lakebase para gravações de transmissão de baixa latência no Lakebase. Este coletor não exige a implementação de funções foreachBatch personalizadas para o gerenciamento de lotes, o gerenciamento de conexão e o tratamento de erros.

Casos de uso comuns incluem:

• Atualize bancos de dados de aplicativos em tempo real para painéis operacionais ou recursos voltados para o cliente.
• Sincronizar dados em constante alteração, como resultados de transmissão agregados ou filtrados, em um banco de dados transacional.
• Grave a saída de uma consulta de transmissão estructurada em uma tabela Lakebase com latência de subsegundo usando modo em tempo real.

Requisitos

• Databricks Runtime 18.3 ou acima
• Compute clássico com modos de acesso Dedicado ou Padrão.
• Um banco de dados Lakebase

Conecte-se a um banco de dados

O coletor do Lakebase oferece suporte aos seguintes métodos de conexão:

Tabelas do Lakebase não registradas no Unity Catalog

Para tabelas do Lakebase não registradas no Unity Catalog, o conector gerencia automaticamente as credenciais e usa a identidade do usuário ou da entidade de serviço do Databricks que executa a consulta.

Para gravar em uma tabela Lakebase, use as opções endpoint, database, dbtable e upsertkey:

Python

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")
  .option("checkpointLocation", "/checkpoints/<query-name>")
  .start()
)

Scala

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")
  .option("checkpointLocation", "/checkpoints/<query-name>")
  .start()

Opções de configuração

O sink gera um erro para opções não reconhecidas, JDBC_STREAMING_SINK_INVALID_OPTIONS.

As seguintes opções se aplicam a todos os métodos de conexão:

Chave	Padrão	Descrição
batchinterval	100 milliseconds	O tempo máximo para manter linhas no buffer antes do esvaziamento. Por exemplo, "50 milliseconds".
batchsize	1000	O número máximo de linhas para cada transação de banco de dados.
checkpointLocation	Nenhuma	Obrigatório. Caminho para o diretório de ponto de verificação.
upsertkey	Nenhuma	Uma lista separada por vírgulas de nomes de colunas que formam a chave de upsert. Por exemplo: "id" ou "user_id,event_type". A tabela de destino deve ter uma restrição PRIMARY KEY nas colunas especificadas. Se não for especificada uma chave de upsert, o coletor infere a chave principal do esquema da tabela de destino. Se não houver uma chave primária, a consulta insere a linha, em vez de atualizar.

Tabelas do Lakebase não registradas com o Unity Catalog

As seguintes opções se aplicam ao se conectar a uma tabela do Lakebase não registrada com o Unity Catalog:

Chave	Padrão	Descrição
database	Nenhuma	O nome da base de dados PostgreSQL de destino.
dbtable	Nenhuma	O nome da tabela de destino no formato schema.table. Se você não especificar um esquema, o valor do esquema default é public.
endpoint	Nenhuma	Especifique o endpoint do Lakebase no formato project_id.branch_id.endpoint_id.

Comportamento de Upsert

Quando existirem chaves de upsert, seja especificadas com upsertkey ou inferidas pelo sink a partir das chaves primárias da tabela, o sink realiza upsert na tabela com a sintaxe INSERT INTO ... ON CONFLICT (<upsert_key>) DO UPDATE SET ... do PostgreSQL.

Quando não existem chaves de upsert, o coletor realiza inserções. O modo de saída da consulta não tem efeito sobre o comportamento de upsert ou insert.

As upsertkey colunas devem:

• Seja um subconjunto não vazio das colunas do DataFrame.
• Faça referência a uma coluna da tabela de destino com uma restrição PRIMARY KEY.
• Devem ser tipos comparáveis, como tipos numéricos ou strings. Para evitar impasses no banco de dados durante gravações concorrentes, o coletor ordena as linhas pela chave de upsert dentro de cada lote. Chaves de upsert não suportam tipos complexos ou struct.

Nomes de coluna são automaticamente citados com o default do PostgreSQL, aspas duplas ", que trata de palavras-chave reservadas, nomes com maiúsculas e minúsculas e caracteres especiais.

O destino não cita os nomes das tabelas e os passa como estão para o banco de dados. Você deve colocar entre aspas os nomes de tabela com caracteres especiais, como "my-schema"."my-table".

Ajuste de desempenho

Processamento em lote e contrapressão

Um flush é acionado quando uma das condições é atendida:

• O buffer atinge batchsize linhas, cujo padrão é 1000.
• A idade do buffer excede batchinterval, que por default é 100 milliseconds.

Quando o banco de dados não consegue acompanhar a taxa de dados de entrada, o destino propaga contrapressão a montante para a fonte.

Latência e Taxa de transferência: Orientações

• Para cargas de trabalho de baixa latência com modo em tempo real, diminua batchinterval para garantir um tempo máximo menor antes da descarga. Consulte modo em tempo real em Transmissão estructurada.
• Para cargas de trabalho de alta taxa de transferência, aumente batchsize para reduzir a sobrecarga de cada transação.

Comportamento da conexão

O sink usa pooling de conexão em executors. Por padrão, cada tarefa usa uma conexão de banco de dados.

O Databricks recomenda que você use o valor default de 1 tarefa para cada conexão. Ao aumentar o número de tarefas para cada conexão, poderá causar contenções de conexão e aumentar as latências para conexões de alta taxa de transferência.

Para configurar a proporção de tarefas para conexões, defina a configuração do Spark spark.databricks.sql.streaming.jdbc.tasksPerConnection. Se o banco de dados de destino tiver um limite de conexão baixo, reduza o número de partições em ordem aleatória ou aumente spark.databricks.sql.streaming.jdbc.tasksPerConnection.

O sink tenta novamente, de forma automática, erros transitórios de JDBC, incluindo falhas de conexão, deadlocks e limitação de taxa. Se o coletor esgotar todas as tentativas, a consulta falhará.

Gatilhos e modos de saída compatíveis

Gatilhos

Esta tabela mostra suporte para tipos de trigger de transmissão estructurada.

Trigger	Suportado
realTime	Sim
ProcessingTime	Sim
AvailableNow	Sim
Once	Sim

Modos de saída

Esta tabela mostra suporte para modos de saída de transmissão estructurada.

Modo de saída	Suportado
update	Sim
append	Sim. O comportamento é idêntico a update. A consulta faz upsert quando a tabela de destino tem uma chave primária, caso contrário, a consulta insere. Ver Comportamento do Upsert.
complete	Não

Limitações

• O compute serverless e Lakeflow Spark Declarative Pipelines não são suportados.
• Somente Lakebase tem suporte como destino de gravação. Bancos de dados externos compatíveis com PostgreSQL não são suportados.