Conectar-se ao Lakebase

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:

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:

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. Veja Conceitos em modo em tempo real para conceitos e exemplos de modo em tempo real para 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:

Modos de saída

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

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.