Conectar-se ao Lakebase

Importante

Esse recurso está em Visualização Pública.

Use o fluxo estruturado para gravar no Lakebase com processamento em lote interno, repetições automáticas e autenticação gerenciada pelo espaço de trabalho.

Quando usar o coletor do Lakebase

Use o coletor do Lakebase para gravações de streaming de baixa latência no Lakebase. Esse coletor não exige que você implemente funções foreachBatch personalizadas para lidar com lotes, gerenciamento de conexões e tratamento de erros.

Os casos de uso comuns incluem:

• Atualize os bancos de dados de aplicativos em tempo real para dashboards operacionais ou recursos voltados para o cliente.
• Sincronizar dados de alteração contínua, como resultados de streaming agregados ou filtrados, em um banco de dados transacional.
• Grave a saída de uma consulta de Structured Streaming em uma tabela do Lakebase com latência inferior a um segundo usando o modo em tempo real.

Para sincronizar dados do Lakebase com tabelas do Delta Lake no Lakehouse, na direção inversa, consulte Lakebase Change Data Feed.

Requisitos

• Databricks Runtime 18 ou superior
• Computação clássica com modos de acesso dedicados ou padrão.
• um banco de dados Lakebase

Conectar a um banco de dados

O coletor lakebase dá suporte aos seguintes métodos de conexão:

Tabelas do Lakebase registradas no Catálogo do Unity

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

Para registrar um banco de dados do Lakebase com o Catálogo do Unity, consulte Registrar um banco de dados lakebase no Catálogo do Unity.

Para gravar em uma tabela do Lakebase, use o método .toTable() com um nome de tabela totalmente qualificado, catalog.schema.table. O exemplo a seguir mostra as opções necessárias, além da opção opcional upsertkey :

Python

(df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")
)

Scala

df.writeStream
  .outputMode("update")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .toTable("<catalog>.<schema>.<table>")

Substitua os seguintes marcadores:

• <catalog>.<schema>.<table>: o nome totalmente qualificado da tabela de destino. O catálogo do Unity Catalog catalog é o catálogo que você criou quando registrou o banco de dados Lakebase; consulte Registrar um banco de dados Lakebase no Unity Catalog. Se a tabela não existir, o conector a criará.
• <primary-key-column>: opcional. Uma lista separada por vírgulas das colunas que formam a chave upsert, por exemplo id ou user_id,event_type. Se você omitir upsertkey, o coletor inferirá a chave da chave primária da tabela de destino. Veja o comportamento de upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: um caminho de volume do Unity Catalog onde a consulta armazena seu ponto de verificação. Você também pode usar um URI de armazenamento de objetos de nuvem. O local deve ser o armazenamento para o qual você pode gravar, não o disco local e deve ser exclusivo para cada consulta de streaming. Isso é independente da tabela de destino. Consulte pontos de verificação do fluxo estruturado.

Para configurações opcionais, como batchsize e batchinterval, consulte as opções de configuração.

Tabelas do Lakebase não registradas no Catálogo do Unity

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

Para gravar em uma tabela Lakebase, use as opções endpoint e dbtable. O exemplo a seguir também inclui as opções opcionais database e upsertkey:

Python

(df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  # Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  # Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()
)

Scala

df.writeStream
  .format("postgresql")
  .outputMode("update")
  .option("endpoint", "<project-id>.<branch-id>.<endpoint-id>")
  .option("database", "<database>")  // Optional. Defaults to databricks_postgres.
  .option("dbtable", "<schema>.<table>")
  .option("upsertkey", "<primary-key-column>")  // Optional. Inferred from the table's primary key if omitted.
  .option("checkpointLocation", "/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>")
  .start()

Substitua os seguintes marcadores:

• <project-id>.<branch-id>.<endpoint-id>: seu endpoint do Lakebase. Localize os três valores no nome do recurso no menu Obter ID da guia Computação , que tem o formato projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Consulte identificadores de computação.
• <database>: opcional. O nome do banco de dados Postgres de destino. Usa databricks_postgres como padrão. Consulte Gerenciar bancos de dados.
• <schema>.<table>: a tabela de destino no formato schema.table. Se você omitir o esquema, o coletor usará o esquema public. Use identificadores simples que comecem com uma letra ou caractere de sublinhado e contenham apenas letras, números e sublinhados; identificadores entre aspas e caracteres especiais, como hífens, não são suportados.
• <primary-key-column>: opcional. Uma lista separada por vírgulas das colunas que formam a chave upsert, por exemplo id ou user_id,event_type. Se você omitir upsertkey, o coletor inferirá a chave da chave primária da tabela de destino. Veja o comportamento de upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: um caminho de volume do Unity Catalog onde a consulta armazena seu ponto de verificação. Você também pode usar um URI de armazenamento de objetos de nuvem. O local deve ser o armazenamento para o qual você pode gravar, não o disco local e deve ser exclusivo para cada consulta de streaming. Isso é independente da tabela de destino. Consulte pontos de verificação do fluxo estruturado.

Para configurações opcionais, como batchsize e batchinterval, consulte as opções de configuração.

Opções de configuração

O coletor retorna um erro ao encontrar 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	Description
batchinterval	100 milliseconds	Optional. O tempo máximo para manter linhas no buffer antes da liberação. Por exemplo, "50 milliseconds".
batchsize	1000	Optional. O número máximo de linhas para cada transação de banco de dados.
checkpointLocation	None	Required. Caminho para um diretório de ponto de verificação, como um volume do Unity Catalog (/Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>). Deve ser exclusivo para cada consulta. Consulte pontos de verificação do fluxo estruturado.
upsertkey	None	Optional. Uma lista separada por vírgulas de nomes de colunas que formam a chave para executar upsert. Por exemplo, "id" ou "user_id,event_type". Se você especificar upsertkey, as colunas deverão corresponder à chave primária da tabela ou a consulta falhará. Se você a omitir, o coletor usará a chave primária automaticamente. Para obter mais informações, consulte Comportamento do Upsert.

Tabelas do Lakebase não registradas no Catálogo do Unity

As seguintes opções se aplicam quando você se conecta a uma tabela do Lakebase não registrada no Catálogo do Unity:

Chave	Padrão	Description
database	databricks_postgres	Optional. O nome do banco de dados PostgreSQL de destino.
dbtable	None	Required. O nome da tabela de destino no formato schema.table. Se você não especificar um esquema, o valor de esquema padrão será public. Use identificadores simples que comecem com uma letra ou um sublinhado e contenham apenas letras, números e sublinhados. Não coloque nomes de tabelas ou esquemas entre aspas; identificadores entre aspas e nomes com caracteres especiais, como hifens, não têm suporte.
endpoint	None	Required. O endpoint do Lakebase, no formato project_id.branch_id ou project_id.branch_id.endpoint_id. O endpoint_id é opcional; se você o omitir e a ramificação tiver um único endpoint de leitura/gravação, o coletor selecionará esse endpoint por padrão.

Comportamento de upsert

Quando existem chaves de upsert, especificadas com upsertkey ou inferidas pelo coletor a partir das chaves primárias da tabela, o coletor executa 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 executa inserções. O modo de saída de uma consulta não afeta o comportamento responsável por executar upsert ou inserção.

As upsertkey colunas devem:

• Seja um subconjunto não vazio das colunas DataFrame.
• Faça a correspondência exata com o PRIMARY KEY da tabela de destino. Se as colunas especificadas não corresponderem à chave primária, a consulta falhará.
• Sejam tipos comparáveis, como tipos numéricos ou de cadeia de caracteres. Para evitar deadlocks no banco de dados durante gravações simultâneas, o coletor classifica as linhas por chave que executa upsert em cada lote. As chaves de upsert não oferecem suporte a tipos complexos ou estruturas.

Os nomes de coluna recebem aspas automaticamente com o padrão do PostgreSQL, aspas duplas ", que lida com palavras-chave reservadas e nomes com maiúsculas e minúsculas.

Os nomes de tabela e esquema devem usar identificadores simples que começam com uma letra ou sublinhado e contêm apenas letras, números e sublinhados. O coletor não dá suporte a identificadores entre aspas nem caracteres especiais, como hifens, em nomes de tabelas ou esquemas.

Ajuste de desempenho

Lotes e contrapressão

Uma descarga é acionada quando uma das condições é atendida:

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

Quando o banco de dados não consegue acompanhar a taxa de dados de entrada, o coletor propaga contrapressão upstream para a origem.

Diretrizes de latência e taxa de transferência:

• Para cargas de trabalho de baixa latência com modo em tempo real, reduza batchinterval para garantir um tempo máximo menor antes da liberação. Consulte o modo em tempo real no Streaming Estruturado para obter conceitos e exemplos de modo em tempo real para obter um exemplo de código.
• Para cargas de trabalho de alta taxa de transferência, aumente batchsize para reduzir a sobrecarga de cada transação.

Comportamento de conexão

O coletor usa pooling de conexões nos executores. Por padrão, cada tarefa usa uma conexão de banco de dados.

O Databricks recomenda que você use o valor padrão da tarefa 1 para cada conexão. Se você 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 baixo de conexões, reduza o número de partições de shuffle ou aumente spark.databricks.sql.streaming.jdbc.tasksPerConnection.

O coletor repete automaticamente erros JDBC transitórios, 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 com suporte

Gatilhos

Esta tabela mostra o suporte para tipos de gatilho de Streaming Estruturado:

Gatilho	Supported
realTime	Yes
ProcessingTime	Yes
AvailableNow	Yes
Once	Yes

Modos de saída

Esta tabela mostra o suporte para modos de saída de Streaming Estruturado:

Modo de saída	Supported
update	Yes
append	Sim. O comportamento é idêntico a update. A consulta executa upsert quando a tabela de destino tem uma chave primária; caso contrário, a consulta insere. Veja o comportamento de upsert.
complete	No

Limitações

• A computação sem servidor e os pipelines do Lakeflow não são compatíveis.
• Somente o Lakebase é compatível como destino de gravação. Não há suporte para bancos de dados compatíveis com PostgreSQL externos.