Conectar-se ao Lakebase

Use o Streaming Estruturado para gravar no Lakebase com processamento em lote integrado, repetições automáticas e autenticação gerenciada pelo workspace.

Quando usar o coletor do Lakebase

Use o coletor do Lakebase para gravações de streaming de baixa latência para o Lakebase. Esse coletor não exige que você implemente funções foreachBatch personalizadas para processar o envio em lote, o gerenciamento de conexões e o 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 com base na chave primária da tabela de destino. Veja o comportamento de upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: um caminho de volume do Catálogo do Unity em que a consulta armazena o 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 de streaming estruturados.

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 ponto de extremidade 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 com base na chave primária da tabela de destino. Veja o comportamento de upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: um caminho de volume do Catálogo do Unity em que a consulta armazena o 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 de streaming estruturados.

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, seja especificadas com upsertkey ou inferidas pelo conector com base nas chaves primárias da tabela, o conector faz 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 tem efeito sobre o comportamento de upsert ou insert.

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 de banco de dados durante gravações simultâneas, o coletor classifica linhas por chave de upsert em cada lote. As chaves de upsert não oferecem suporte a tipos complexos ou estruturas.

Os nomes de coluna são citados automaticamente usando o padrão do PostgreSQL, aspas duplas ", que trata palavras-chave reservadas e nomes que usam 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 ou caracteres especiais, como hifens, em nomes de tabela ou esquema.

Ajuste de desempenho

Envio em lote e contrapressão

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

• O buffer chega a linhas batchsize, com o 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 a contrapressão upstream até 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 para cada transação.

Comportamento de conexão

O coletor usa o pool de conexões em 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 de conexão baixo, reduza o número de partições aleatórias ou aumente spark.databricks.sql.streaming.jdbc.tasksPerConnection.

O coletor repete automaticamente erros JDBC temporá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

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