Liga-se a Lakebase

Utilize o Structured Streaming para escrever no Lakebase com processamento em lotes incorporado, novas tentativas automáticas e autenticação gerida pela área de trabalho.

Quando usar o sumidouro Lakebase

Utilize o sink do Lakebase para escritas em fluxo de baixa latência no Lakebase. Este sink não exige que implementes funções personalizadas foreachBatch para gerir batching, gestão de ligações e gestão de erros.

Os casos de uso comuns incluem:

• Atualize as bases de dados das aplicações em tempo real para dashboards operacionais ou funcionalidades voltadas para o cliente.
• Sincronizar dados em constante mudança, como resultados agregados ou filtrados de streaming, numa base de dados transacional.
• Escreva a saída de uma consulta de Streaming Estruturado numa tabela Lakebase com latência inferior a um segundo usando o modo em tempo real.

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

Requisitos

• Databricks Runtime 18 e superiores
• Computação clássica com modos de acesso dedicados ou padrão.
• Uma base de dados Lakebase

Ligar a uma base de dados

O sumidouro Lakebase suporta os seguintes métodos de ligação:

Tabelas Lakebase registadas no Unity Catalog

Para tabelas Lakebase registadas no Unity Catalog, o conector gere automaticamente as credenciais e utiliza a identidade do utilizador ou do principal do serviço que executa a consulta. Se a tabela não existir, o conector cria a tabela.

Para registar uma base de dados Lakebase no Unity Catalog, consulte Registar uma base de dados Lakebase no Unity Catalog.

Para escrever numa tabela Lakebase, use o .toTable() método com um nome de tabela totalmente qualificado, catalog.schema.table. O exemplo seguinte mostra as opções obrigatórias, mais a 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 placeholders:

• <catalog>.<schema>.<table>: O nome totalmente qualificado da tabela alvo. Este catalog é o catálogo do Catálogo Unity que criou quando registou a base de dados Lakebase, veja Registar uma base de dados Lakebase no Catálogo Unity. Se a tabela não existir, o conector cria-a.
• <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 omitir upsertkey, o sumidouro infere a chave a partir da chave primária da tabela alvo. Ver comportamento de Upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Um caminho para um volume do Unity Catalog onde a consulta armazena o ponto de verificação. Também pode usar um URI de armazenamento de objetos na cloud. A localização deve ser armazenamento onde possas escrever, não disco local, e deve ser única para cada consulta de streaming. Isto é independente da tabela alvo. Consulte Pontos de verificação de streaming estruturado.

Para configurações opcionais, como batchsize e batchinterval, veja Opções de configuração.

Tabelas Lakebase não registadas no Unity Catalog

Para tabelas Lakebase não registadas no Unity Catalog, o conector gere automaticamente as credenciais e utiliza a identidade do utilizador ou principal do serviço que executa a consulta. Se a tabela não existir, o conector cria a tabela.

Para escrever numa tabela do Lakebase, use as opções endpoint e dbtable. O exemplo seguinte inclui também 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 placeholders:

• <project-id>.<branch-id>.<endpoint-id>: O seu endpoint Lakebase. Encontre os três valores no nome do Recurso no menu Obter ID do separador Computes , que tem o formato projects/<project-id>/branches/<branch-id>/endpoints/<endpoint-id>. Ver Identificadores de computação.
• <database>: Opcional. O nome da base de dados Postgres alvo. O valor padrão é databricks_postgres. Consulte Gerir bases de dados.
• <schema>.<table>: A tabela alvo em schema.table formato. Se omitir o esquema, o sink utiliza o esquema public. Use identificadores simples que comecem por uma letra ou sublinhado e contenham apenas letras, números e sublinhados; identificadores delimitados por 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 omitir upsertkey, o sumidouro infere a chave a partir da chave primária da tabela alvo. Ver comportamento de Upsert.
• /Volumes/<catalog>/<schema>/<volume>/<checkpoint-name>: Um caminho para um volume do Unity Catalog onde a consulta armazena o ponto de verificação. Também pode usar um URI de armazenamento de objetos na cloud. A localização deve ser armazenamento onde possas escrever, não disco local, e deve ser única para cada consulta de streaming. Isto é independente da tabela alvo. Consulte Pontos de verificação de streaming estruturado.

Para configurações opcionais, como batchsize e batchinterval, veja Opções de configuração.

Opções de configuração

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

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

Tabelas Lakebase não registadas no Unity Catalog

As seguintes opções aplicam-se quando se liga a uma tabela Lakebase não registada no Unity Catalog:

Comportamento de Upsert

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

Quando não existem chaves de upsert, o sink efetua inserções. O modo de saída de uma consulta não afeta o comportamento do upsert nem o comportamento de inserção.

As upsertkey colunas devem:

• Ser um subconjunto não vazio das colunas DataFrame.
• Corresponde exatamente à tabela PRIMARY KEY de alvo. Se as colunas que especificas não corresponderem à chave primária, a consulta falha.
• Sejam tipos comparáveis, como números ou tipos de cadeia. Para evitar bloqueios da base de dados durante escritas concorrentes, o sink ordena as linhas por chave upsert dentro de cada lote. As chaves upsert não suportam tipos complexos ou tipos estruturados.

Os nomes das colunas são automaticamente delimitados pelas aspas duplas predefinidas do PostgreSQL ", o que permite lidar com palavras-chave reservadas e nomes com mistura de maiúsculas e minúsculas.

Os nomes de tabelas e esquemas devem usar identificadores simples que começam com uma letra ou sublinhado e contêm apenas letras, números e sublinhados. O sumidouro não suporta identificadores entre aspas nem caracteres especiais, como hífens, em nomes de tabelas ou esquemas.

Afinação de Performance

Processamento em lote e retropressão

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

• O buffer atinge batchsize linhas, sendo o valor predefinido 1000.
• A antiguidade do buffer excede batchinterval, cujo valor por defeito é 100 milliseconds.

Quando a base de dados não consegue acompanhar a taxa de dados recebidos, o sumidouro propaga a contrapressão a montante até à fonte.

Orientação sobre latência e débito:

• Para cargas de trabalho de baixa latência com modo de tempo real, diminua batchinterval para garantir um tempo máximo mais curto antes da descarga. Veja Modo em tempo real em Structured Streaming para conceitos e exemplos de modo em tempo real para um exemplo de código.
• Para cargas de trabalho de alto rendimento, aumente batchsize para reduzir a sobrecarga de cada transação.

Comportamento de ligação

O sink utiliza agrupamento de ligações nos executores. Por defeito, cada tarefa utiliza uma ligação à base de dados.

A Databricks recomenda que utilize o valor predefinido de 1 para cada ligação. Se aumentares o número de tarefas para cada ligação, podes causar conflitos na ligação e aumentar as latências para ligações de alto débito.

Para configurar a proporção de tarefas para ligações, defina a spark.databricks.sql.streaming.jdbc.tasksPerConnection configuração do Spark. Se a base de dados alvo tiver um limite baixo de ligações, reduza o número de partições de mistura ou aumente spark.databricks.sql.streaming.jdbc.tasksPerConnection.

O sink tenta automaticamente erros JDBC transitórios, incluindo falhas de ligação, deadlocks e limitação de taxa. Se o sink esgotar o número máximo de tentativas, a consulta falha.

Acionadores suportados e modos de saída

Triggers

Esta tabela mostra suporte para tipos de gatilhos 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 o Lakeflow Spark Declarative Pipelines não são suportados.
• Apenas o Lakebase é suportado como destino de escrita. Bases de dados externas compatíveis com PostgreSQL não são suportadas.