Servir dados do Lakehouse com tabelas sincronizadas

As tabelas sincronizadas permitem que você forneça dados do Lakehouse por meio do Lakebase Postgres. As tabelas do Catálogo do Unity são sincronizadas com o Postgres para que os aplicativos possam consultar dados do Lakehouse diretamente com baixa latência. Esse processo é comumente conhecido como ETL reverso. O lakehouse é otimizado para análise e enriquecimento, enquanto o Lakebase foi projetado para cargas de trabalho operacionais que exigem consultas de estilo de pesquisa rápida e consistência transacional.

O que são tabelas sincronizadas?

As tabelas sincronizadas permitem que você atenda dados de nível de análise do Catálogo do Unity por meio do Lakebase Postgres, disponibilizando-os para aplicativos que precisam de consultas de baixa latência e transações ACID completas. Eles fazem a ponte entre o armazenamento analítico e os sistemas operacionais, mantendo seus dados prontos para servir em aplicativos em tempo real.

Fontes Suportadas

Tabelas sincronizadas suportam os seguintes tipos de origem do Unity Catalog:

• Tabelas Delta gerenciadas e externas
• Tabelas de Iceberg gerenciadas e externas
• Visões e visões materializadas

Como funciona

As tabelas sincronizadas do Databricks criam uma cópia gerenciada dos dados do Catálogo do Unity no Lakebase. Ao criar uma tabela sincronizada, você obtém:

1. Uma tabela sincronizada no Catálogo do Unity que faz referência ao pipeline de sincronização
2. Uma tabela postgres no Lakebase (somente leitura, que pode ser consultada por seus aplicativos)

Por exemplo, você pode sincronizar tabelas de ouro, recursos projetados ou saídas de ML de analytics.gold.user_profiles em uma nova tabela analytics.gold.user_profiles_synced sincronizada. No Postgres, o nome do esquema do Catálogo do Unity se torna o nome do esquema Postgres, portanto, isso aparece como gold.user_profiles_synced:

SELECT * FROM gold.user_profiles_synced WHERE user_id = 12345;

Os aplicativos se conectam com drivers Postgres padrão, e consultam os dados sincronizados junto com seu próprio estado operacional.

Os pipelines de sincronização usam pipelines gerenciados e declarativos do Spark do Lakeflow para atualizar continuamente tanto a tabela sincronizada do Unity Catalog quanto a tabela Postgres com as mudanças da tabela de origem. Cada sincronização pode usar até 16 conexões com seu banco de dados lakebase.

O Lakebase Postgres dá suporte a até 1.000 conexões simultâneas com garantias transacionais, para que os aplicativos possam ler dados enriquecidos ao mesmo tempo em que manipulam inserções, atualizações e exclusões no mesmo banco de dados.

Modos de sincronização

Escolha o modo de sincronização correto com base nas necessidades do aplicativo:

Os modos disparado e contínuo exigem que a alimentação de dados de alteração (CDF) esteja habilitada na tabela de origem. Se o CDF não estiver habilitado, você verá um aviso na interface do usuário com o comando exato ALTER TABLE a ser executado. Para obter mais detalhes sobre o Feed de Dados de Alterações, consulte Uso do feed de dados de alterações do Delta Lake no Databricks.

Exemplos de casos de uso

Você pode usar tabelas sincronizadas para casos de uso de serviço de dados, como:

• Mecanismos de personalização que atendem novos perfis de usuário aos Aplicativos do Databricks
• Aplicativos que fornecem previsões de modelo ou valores de características computados no lakehouse
• Painéis voltados para o cliente que atendem KPIs em tempo real
• Serviços de detecção de fraudes que fornecem pontuações de risco para ações imediatas
• Ferramentas de suporte que fornecem registros de clientes enriquecidos a partir de dados do lakehouse

Criar uma tabela sincronizada

Pré-requisitos

Você precisa de:

• Um workspace do Databricks com o Lakebase habilitado.
• Um projeto lakebase (consulte Criar um projeto).
• Uma tabela do Unity Catalog para sincronização.
• Permissões para criar tabelas sincronizadas. Você precisa USE_SCHEMA e CREATE_TABLE em qualquer esquema usado.

Para os modos Triggered ou Continuous, o Feed de Dados de Alteração deve estar habilitado na tabela de origem.

ALTER TABLE your_catalog.your_schema.your_table
SET TBLPROPERTIES (delta.enableChangeDataFeed = true)

Para planejamento de capacidade e compatibilidade de tipo de dados, consulte Tipos de dados e compatibilidade e planejamento de capacidade.

interface do usuário

1. Vá para Catálogo na barra lateral do workspace e selecione a tabela catálogo do Unity que você deseja sincronizar.

   

2. Clique em Criar>tabela sincronizada na exibição de detalhes da tabela.

   

3. Na caixa de diálogo Criar tabela sincronizada :

   As listas de catálogo e esquema incluem apenas esquemas do Catálogo do Unity em que o usuário atual tem privilégios de USE_SCHEMA e CREATE_TABLE. Se você não vir o esquema que esperava, confirme suas permissões com o administrador do catálogo.

   1. Nome da tabela: insira um nome para sua tabela sincronizada (ela é criada no mesmo catálogo e esquema que a tabela de origem). Isso cria uma tabela sincronizada do Catálogo do Unity e uma tabela postgres que você pode consultar.

   2. Tipo de banco de dados: Escolha Lakebase Serverless (Dimensionamento Automático).

   3. Modo de sincronização: escolha Instantâneo, Acionado ou Contínuo com base em suas necessidades (confira os modos de sincronização acima).

   4. Configure suas seleções de projeto, branch e banco de dados.

   5. Verifique se a chave primária está correta (geralmente detectada automaticamente).

      Importante

      As colunas na chave primária não são anuláveis na tabela sincronizada. Linhas com nulos em colunas de chave primária são excluídas da sincronização.

   6. (Opcional) Se duas linhas puderem compartilhar a mesma chave primária na tabela de origem, selecione uma chave Timeseries para configurar a eliminação de duplicação. Quando uma chave de timeseries é especificada, a tabela sincronizada contém apenas a linha com o valor da chave de timeseries mais recente para cada chave primária. Para o modo de falha sem uma chave de timeseries, consulte Chaves duplicadas.

   Se você escolheu o modo Disparado ou o modo Contínuo e ainda não habilitou o Feed de Alterações de Dados, verá um aviso com o comando exato a ser executado. Para perguntas de compatibilidade de tipo de dados, consulte Tipos de dados e compatibilidade.

   Clique em Criar para criar a tabela sincronizada.

4. Monitore a tabela sincronizada no Catálogo. A guia Visão geral mostra o status da sincronização, a configuração, o status do pipeline e o carimbo de data/hora da última sincronização. Use Sincronizar agora para atualização imediata.

SDK do Python

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.postgres import (
    SyncedTable,
    SyncedTableSyncedTableSpec,
    SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy,
)

w = WorkspaceClient()

synced_table = w.postgres.create_synced_table(
    synced_table=SyncedTable(spec=SyncedTableSyncedTableSpec(
        source_table_full_name="main.sales.orders",
        branch="projects/my-project/branches/production",
        primary_key_columns=["order_id"],
        scheduling_policy=SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy.SNAPSHOT,
        postgres_database="mydb",
        create_database_objects_if_missing=True,
    )),
    synced_table_id="my-catalog.sales.orders",
).wait()

print(f"Synced table created: {synced_table.name}")

A entidade synced_table_id utiliza o formato catalog.schema.table e transforma-se no nome da tabela sincronizada do Unity Catalog. No Postgres, a tabela {table} é criada no esquema {schema}, dentro do banco de dados com postgres_database o qual você definiu (aqui, mydb).

SDK do Java

import com.databricks.sdk.WorkspaceClient;
import com.databricks.sdk.service.postgres.*;
import java.util.List;

WorkspaceClient w = new WorkspaceClient();

SyncedTable syncedTable = w.postgres().createSyncedTable(
    new CreateSyncedTableRequest()
        .setSyncedTableId("my-catalog.sales.orders")
        .setSyncedTable(new SyncedTable()
            .setSpec(new SyncedTableSyncedTableSpec()
                .setSourceTableFullName("main.sales.orders")
                .setBranch("projects/my-project/branches/production")
                .setPrimaryKeyColumns(List.of("order_id"))
                .setSchedulingPolicy(SyncedTableSyncedTableSpecSyncedTableSchedulingPolicy.SNAPSHOT)
                .setPostgresDatabase("mydb")
                .setCreateDatabaseObjectsIfMissing(true))))
    .waitForCompletion();

System.out.println("Synced table created: " + syncedTable.getName());

encurvar

curl -X POST "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables?synced_table_id=my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
  -H "Content-Type: application/json" \
  -d '{
    "spec": {
      "source_table_full_name": "main.sales.orders",
      "branch": "projects/my-project/branches/production",
      "primary_key_columns": ["order_id"],
      "scheduling_policy": "SNAPSHOT",
      "postgres_database": "mydb",
      "create_database_objects_if_missing": true
    }
  }'

Isso retorna uma operação de longo prazo de execução. Sondar o campo retornado name até done: true. Consulte operações de longa duração. Para configurar a autenticação, consulte Autenticação.

Agendar ou iniciar sincronizações subsequentes

O instantâneo inicial é executado automaticamente na criação. Para os modos Instantâneo e Disparado, as sincronizações subsequentes devem ser disparadas explicitamente. O modo contínuo se gerencia automaticamente.

Tarefa de pipeline de sincronização de tabelas do banco de dados

A tarefa Database Table Sync pipeline nos Lakeflow Jobs executa o pipeline de uma tabela sincronizada como uma etapa do fluxo de trabalho. Configure o trabalho com um gatilho de atualização de tabela ou um agendamento.

Gatilho para atualizações na tabela de origem

Aciona o trabalho quando a tabela de origem do Unity Catalog é atualizada. Com o modo Acionado, apenas novas alterações são aplicadas de forma incremental, proporcionando atualizações quase em tempo real sem o custo sempre ativo do modo Contínuo.

1. Na barra lateral, clique em Fluxos de Trabalho.
2. Clique em Criar trabalho ou abra um trabalho existente.
3. Na guia Tarefas , clique em + Adicionar outro tipo de tarefa.
4. Em Ingestão e Transformação, selecione pipeline de Sincronização de Tabela de Banco de Dados.
5. No campo Pipeline, selecione o pipeline associado à sua tabela sincronizada.
6. Em Agendas & Gatilhos, clique em Adicionar gatilho.
7. Selecione atualização da tabela como o tipo de gatilho.
8. Em Tabelas, selecione a tabela de origem do Catálogo do Unity a ser monitorada.
9. Clique em Salvar.

Disparar em um agendamento

Executa a sincronização em uma cadência fixa. Adequado para o modo Snapshot, em que uma atualização completa noturna ou semanal geralmente é o padrão mais eficiente.

1. Siga as etapas 1 a 5 acima para adicionar uma tarefa de pipeline de Sincronização de Tabela de Banco de Dados a um trabalho.
2. Em Agendas & Gatilhos, clique em Adicionar gatilho.
3. Selecione Agendado como tipo de gatilho.
4. Defina o agendamento cron e o fuso horário e clique em Salvar.

Verificar o status da sincronização

Para verificar o estado atual e a hora da última sincronização de uma tabela sincronizada:

interface do usuário

No Catálogo, navegue até a tabela sincronizada e selecione a guia Visão geral . Ele mostra o estado de sincronização atual, o status do pipeline e o carimbo de data/hora da última sincronização.

SDK do Python

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

table = w.postgres.get_synced_table("synced_tables/my-catalog.sales.orders")
print(f"State: {table.status.detailed_state}")
print(f"Last sync: {table.status.last_sync_time}")
print(f"Message: {table.status.message}")

SDK do Java

import com.databricks.sdk.WorkspaceClient;
import com.databricks.sdk.service.postgres.SyncedTable;

WorkspaceClient w = new WorkspaceClient();

SyncedTable table = w.postgres().getSyncedTable("synced_tables/my-catalog.sales.orders");
System.out.println("State: " + table.getStatus().getDetailedState());
System.out.println("Last sync: " + table.getStatus().getLastSyncTime());
System.out.println("Message: " + table.getStatus().getMessage());

encurvar

curl "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables/my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

Tipos de dados e compatibilidade

Os tipos de dados do Catálogo do Unity são mapeados para tipos Postgres ao criar tabelas sincronizadas. Tipos complexos (ARRAY, MAP, STRUCT) são armazenados como JSONB no Postgres.

Manipular caracteres inválidos

Determinados caracteres, como bytes nulos (0x00), são permitidos nas colunas STRING, ARRAY, MAP ou STRUCT do Unity, mas não têm suporte em colunas POSTGRES TEXT ou JSONB. Isso pode causar falhas de sincronização com erros como:

ERROR: invalid byte sequence for encoding "UTF8": 0x00
ERROR: unsupported Unicode escape sequence DETAIL: \u0000 cannot be converted to text

• O primeiro erro ocorre quando um byte nulo aparece em uma coluna de cadeia de caracteres de nível superior, que é mapeada diretamente para Postgres TEXT.
• O segundo erro ocorre quando um byte nulo aparece em uma cadeia de caracteres aninhada dentro de um tipo complexo (STRUCTou ARRAYMAP), que é serializado como JSONB. Durante a serialização, todas as cadeias de caracteres são convertidas em Postgres TEXT, onde \u0000 não é permitido.

Soluções:

• Sanitizar campos de cadeia de caracteres: remova caracteres sem suporte antes de sincronizar. Para bytes nulos em colunas STRING:

  SELECT REPLACE(column_name, CAST(CHAR(0) AS STRING), '') AS cleaned_column FROM your_table
  

• Converter para BINARY: para colunas STRING em que a preservação de bytes brutos é necessária, converter para tipo BINARY.

Planejamento de capacidade

Ao planejar a implementação de tabelas sincronizadas, considere estes requisitos de recurso:

• Uso da conexão: cada tabela sincronizada usa até 16 conexões com seu banco de dados lakebase, que contam para o limite de conexão da instância.
• Limites de tamanho: o limite total de tamanho de dados lógicos em todas as tabelas sincronizadas é de 8 TB. Tabelas individuais não têm limites, mas o Databricks recomenda não exceder 1 TB para tabelas que exigem atualizações.
• Tamanho da atualização completa: ao disparar uma atualização completa, a versão antiga no Postgres não será excluída até que a nova sincronização seja concluída. Ambas as versões contam temporariamente para o limite de tamanho do banco de dados lógico durante a atualização.
• Tabelas por origem: uma única tabela de origem pode ter até 20 tabelas sincronizadas.
• Requisitos de nomenclatura: os nomes de banco de dados, esquema e tabela podem conter apenas caracteres alfanuméricos e sublinhados ([A-Za-z0-9_]+).
• Orientações sobre o identificador de origem: evite usar letras maiúsculas ou caracteres especiais em nomes de colunas ou tabelas na tabela de origem do Unity Catalog. Se você optar por mantê-los, deverá colocar esses identificadores entre aspas ao se referir a eles no Postgres.
• Evolução do esquema: há suporte apenas para alterações de esquema aditivo (como adicionar colunas) para modos disparados e contínuos.
• Chaves duplicadas: se duas linhas tiverem a mesma chave primária na tabela de origem, o pipeline de sincronização falhará, a menos que você configure a eliminação de duplicação usando uma chave de timeseries.
• Idempotência da API: As APIs de tabelas sincronizadas são idempotentes, portanto, em caso de erros transitórios, tente novamente para garantir operações em tempo hábil.
• Taxa de atualização: Para o dimensionamento automático do Lakebase, o pipeline de sincronização oferece suporte a gravações contínuas e acionadas a aproximadamente 150 linhas por segundo por Unidade de Capacidade (CU) e a gravações de instantâneo de até 2.000 linhas por segundo por CU.

Operações permitidas em tabelas sincronizadas no Postgres

Azure Databricks recomenda executar apenas as seguintes operações no Postgres para tabelas sincronizadas para evitar substituições acidentais ou inconsistências de dados:

• Consultas somente leitura
• Criando índices
• Soltando a tabela (para liberar espaço depois de remover a tabela sincronizada do Catálogo do Unity)

Embora seja possível modificar tabelas sincronizadas no Postgres de outras maneiras, ele interfere no pipeline de sincronização.

Propriedade e permissões

Se você criar um novo banco de dados, esquema ou tabela do Postgres, a propriedade do Postgres será definida da seguinte maneira:

• A propriedade é atribuída ao usuário que cria o banco de dados, esquema ou tabela, se o login Azure Databricks existir como uma função no Postgres. Para adicionar uma função de identidade Azure Databricks no Postgres, consulte Postgres roles.
• Caso contrário, a propriedade é atribuída ao proprietário do objeto pai no Postgres (normalmente o databricks_superuser).

Gerenciar o acesso à tabela sincronizada

Depois que uma tabela sincronizada é criada, databricks_superuser pode ler uma tabela sincronizada do Postgres. O databricks_superuser possui pg_read_all_data que permite que essa função leia de todas as tabelas. Ele também tem o pg_write_all_data privilégio, que permite que essa função escreva em todas as tabelas. Isso significa que um databricks_superuser também pode gravar em uma tabela sincronizada no Postgres. O Lakebase dá suporte a esse comportamento de gravação caso você precise fazer alterações urgentes em sua tabela de destino. No entanto, o Azure Databricks recomenda corrigir a tabela de origem em vez disso.

• O databricks_superuser também pode conceder esses privilégios a outros usuários:

  GRANT USAGE ON SCHEMA synced_table_schema TO user;
  

  GRANT SELECT ON synced_table_name TO user;
  

• O databricks_superuser pode revogar esses privilégios:

  REVOKE USAGE ON SCHEMA synced_table_schema FROM user;
  

  REVOKE {SELECT | INSERT | UPDATE | DELETE} ON synced_table_name FROM user;
  

Gerenciar operações de tabela sincronizadas

O databricks_superuser pode gerenciar quais usuários estão autorizados a executar operações específicas em uma tabela sincronizada. As operações com suporte para tabelas sincronizadas são:

• CREATE INDEX
• ALTER INDEX
• DROP INDEX
• DROP TABLE

Todas as outras operações DDL são negadas para tabelas sincronizadas.

Para conceder esses privilégios a usuários adicionais, primeiro databricks_superuser deve criar uma extensão em databricks_auth:

CREATE EXTENSION IF NOT EXISTS databricks_auth;

Em seguida, o databricks_superuser pode adicionar um usuário para gerenciar uma tabela sincronizada:

SELECT databricks_synced_table_add_manager('"synced_table_schema"."synced_table"'::regclass, '[user]');

databricks_superuser pode remover um usuário do gerenciamento de uma tabela sincronizada.

SELECT databricks_synced_table_remove_manager('[table]', '[user]');

O databricks_superuser pode ver todos os gerentes:

SELECT * FROM databricks_synced_table_managers;

Excluir uma tabela sincronizada

Excluir uma tabela sincronizada do Catálogo do Unity também descarta a tabela Postgres correspondente.

interface do usuário

No Catálogo, localize a tabela sincronizada, clique no Menu e selecione Excluir.

SDK do Python

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

w.postgres.delete_synced_table("synced_tables/my-catalog.sales.orders").wait()

SDK do Java

import com.databricks.sdk.WorkspaceClient;

WorkspaceClient w = new WorkspaceClient();

w.postgres().deleteSyncedTable("synced_tables/my-catalog.sales.orders").waitForCompletion();

encurvar

curl -X DELETE "https://your-workspace.cloud.databricks.com/api/2.0/postgres/synced_tables/my-catalog.sales.orders" \
  -H "Authorization: Bearer ${DATABRICKS_TOKEN}"

Saiba mais

Integração de catálogo

• Duplicação de catálogo: A criação de uma tabela sincronizada em um catálogo padrão direcionado a um banco de dados Postgres, que também está registrado como um catálogo de banco de dados separado, faz com que a tabela sincronizada apareça no Unity Catalog sob ambos os catálogos, padrão e de banco de dados.

Outras opções

Para sincronizar dados em sistemas que não são do Databricks, consulte soluções de Reverse ETL do Partner Connect, como Census ou Hightouch.