Erogare dati lakehouse con tabelle sincronizzate (Lakebase provisionato)

Le tabelle sincronizzate consentono di gestire i dati lakehouse tramite le istanze di Postgres con provisioning di Lakebase. Le tabelle del catalogo Unity vengono sincronizzate in Postgres in modo che le applicazioni possano eseguire query sui dati lakehouse direttamente con bassa latenza. Questo processo è comunemente noto come ETL inverso. Il lakehouse è ottimizzato per l'analisi e l'arricchimento, mentre Lakebase è progettato per carichi di lavoro operativi che richiedono query in stile ricerca veloce e coerenza transazionale.

Che cosa sono le tabelle sincronizzate?

Le tabelle sincronizzate consentono di gestire dati di livello di analisi da Unity Catalog tramite Lakebase Provisioned Postgres, rendendoli disponibili per le applicazioni che necessitano di query a bassa latenza e transazioni ACID complete. Consentono di colmare il divario tra l'archiviazione analitica e i sistemi operativi mantenendo i dati pronti per la gestione in applicazioni in tempo reale.

Le pipeline di sincronizzazione utilizzano le Lakeflow Spark Declarative Pipelines gestite per aggiornare continuamente sia la tabella sincronizzata di Unity Catalog che la tabella Postgres con le modifiche dalla tabella di origine. Dopo la creazione, è possibile eseguire query sulle tabelle sincronizzate direttamente usando gli strumenti Postgres.

Le caratteristiche principali delle tabelle sincronizzate sono le seguenti:

• La tabella viene considerata di sola lettura in Postgres per mantenere l'integrità dei dati con l'origine
• Sincronizzazione automatica con le pipeline dichiarative di Spark di Lakeflow gestite
• Interrogabile tramite interfacce standard PostgreSQL
• Gestito tramite il catalogo Unity per la governance e la gestione del ciclo di vita

Prima di iniziare

• È disponibile una tabella del catalogo Unity in qualsiasi catalogo.
• Hai CAN USE autorizzazioni sull'istanza del database.

Creare una tabella sincronizzata

INTERFACCIA UTENTE

Per sincronizzare una tabella del catalogo Unity in Postgres, eseguire le operazioni seguenti:

1. Fare clic su Catalogo nella barra laterale dell'area di lavoro.

2. Trovare e selezionare la tabella del catalogo Unity in cui si vuole creare una tabella sincronizzata.

3. Fare clic su Crea>tabella sincronizzata.

4. Selezionare il catalogo, lo schema e immettere un nome di tabella per la nuova tabella sincronizzata.

   • Le tabelle sincronizzate possono essere create anche nei cataloghi Standard, con alcune configurazioni aggiuntive. Selezionare il catalogo Standard, uno schema e immettere un nome di tabella per la tabella sincronizzata appena creata.

5. Selezionare un'istanza del database e immettere il nome del database Postgres in cui creare la tabella sincronizzata. Per impostazione predefinita, il campo del database Postgres corrisponde al catalogo di destinazione attualmente selezionato. Se un database Postgres non esiste con questo nome, Azure Databricks ne crea uno nuovo.

6. Selezionare una chiave primaria. Una chiave primaria è necessaria perché consente un accesso efficiente alle righe per letture, aggiornamenti ed eliminazioni.

   Importante

   Le colonne nella chiave primaria non sono annullabili nella tabella sincronizzata. Di conseguenza, le righe con valori Null nelle colonne chiave primaria vengono escluse dalla sincronizzazione.

7. Se due righe hanno la stessa chiave primaria nella tabella di origine, selezionare una chiave Timeseries per configurare la deduplicazione. Quando viene specificata una chiave timeseries , le tabelle sincronizzate contengono solo le righe con il valore della chiave timeseries più recente per ogni chiave primaria.

8. Selezionare la modalità di sincronizzazione da Snapshot, Attivata e Continua. Per altre informazioni su ogni modalità di sincronizzazione, vedere Spiegazione delle modalità di sincronizzazione.

9. Scegliere se si vuole creare questa tabella sincronizzata da una pipeline nuova o esistente.

   • Se si crea una nuova pipeline e si utilizza un catalogo gestito, scegliere la posizione di archiviazione per la tabella di staging. Se si usa un catalogo standard, la tabella di staging viene archiviata automaticamente nel catalogo.
   • Se si usa una pipeline esistente, verificare che la nuova modalità di sincronizzazione corrisponda alla modalità pipeline.

10. (Facoltativo) Selezionare un Serverless Budget Policy. Per creare una politica di budget serverless, vedere Utilizzo degli attributi con politiche di budget serverless. In questo modo è possibile attribuire l'utilizzo della fatturazione a criteri di utilizzo specifici.

    • Per le tabelle sincronizzate, l'entità fatturabile è la pipeline dichiarativa sottostante di Lakeflow Spark. Per modificare i criteri di budget, modificare l'oggetto pipeline sottostante. Consulta Configurare una pipeline serverless.

11. Dopo che lo stato della tabella sincronizzata è Online, accedere all'istanza del database ed eseguire una query sulla tabella appena creata. Esegui query sulla tabella utilizzando l'editor SQL, strumenti esterni o notebook.

PYTHON SDK

from databricks.sdk import WorkspaceClient
from databricks.sdk.service.database import SyncedDatabaseTable, SyncedTableSpec, NewPipelineSpec, SyncedTableSchedulingPolicy

# Initialize the Workspace client
w = WorkspaceClient()

# Create a synced table in a database catalog
synced_table = w.database.create_synced_database_table(
    SyncedDatabaseTable(
        name="database_catalog.schema.synced_table",  # Full three-part name
        spec=SyncedTableSpec(
            source_table_full_name="source_catalog.source_schema.source_table",
            primary_key_columns=["id"],  # Primary key columns
            scheduling_policy=SyncedTableSchedulingPolicy.TRIGGERED,  # SNAPSHOT, TRIGGERED, or CONTINUOUS
            # Optional: timeseries_key="timestamp"  # For deduplication
            new_pipeline_spec=NewPipelineSpec(
                storage_catalog="storage_catalog",
                storage_schema="storage_schema"
            )
        ),
    )
)
print(f"Created synced table: {synced_table.name}")

# Create a synced table in a standard UC catalog
synced_table = w.database.create_synced_database_table(
    SyncedDatabaseTable(
        name="standard_catalog.schema.synced_table",  # Full three-part name
        database_instance_name="my-database-instance",  # Required for standard catalogs
        logical_database_name="postgres_database",  # Required for standard catalogs
        spec=SyncedTableSpec(
            source_table_full_name="source_catalog.source_schema.source_table",
            primary_key_columns=["id"],
            scheduling_policy=SyncedTableSchedulingPolicy.CONTINUOUS,
            create_database_objects_if_missing=True,  # Create database/schema if needed
            new_pipeline_spec=NewPipelineSpec(
                storage_catalog="storage_catalog",
                storage_schema="storage_schema"
            )
        ),
    )
)
print(f"Created synced table: {synced_table.name}")

# Check the status of a synced table
synced_table_name = "database_catalog.schema.synced_table"
status = w.database.get_synced_database_table(name=synced_table_name)
print(f"Synced table status: {status.data_synchronization_status.detailed_state}")
print(f"Status message: {status.data_synchronization_status.message}")

CLI

# Create a synced table in a database catalog
databricks database create-synced-database-table  \
  --json '{
    "spec": {
      "name": "database_catalog.schema.synced_table",
      "source_table_full_name": "source_catalog.source_schema.source_table",
      "primary_key_columns": ["id"],
      "scheduling_policy": "TRIGGERED"
    },
    "new_pipeline_spec": {
      "storage_catalog": "storage_catalog",
      "storage_schema": "storage_schema"
    }
  }'

# Create a synced table in a standard UC catalog
# new_pipeline_spec, storage_catalog, and storage_schema are optional
databricks database create-synced-database-table \
  --database-instance-name "my-database-instance" \
  --logical-database-name "databricks_postgres" \
  --json '{
    "name": "standard_catalog.schema.synced_table",
    "spec": {
      "source_table_full_name": "source_catalog.source_schema.source_table",
      "primary_key_columns": ["id"],
      "scheduling_policy": "CONTINUOUS",
      "create_database_objects_if_missing": true
    }
  }'

# Check the status of a synced table
databricks database get-synced-database-table "database_catalog.schema.synced_table"

curva

Creare una tabella sincronizzata in un catalogo di database.

export CATALOG_NAME=<Database catalog>
export SRC_TBL="source_catalog.source_schema.source_table"
export DEST_TBL="$CATALOG_NAME.some_schema.synced_table"
export PKS='["id"]'
export ST_CATALOG="storage_catalog"
export ST_SCHEMA="storage_schema"

curl -X POST https://$WORKSPACE/api/2.0/database/synced_tables \
-H "Content-Type: text/json" \
-H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
--data-binary @- << EOF
{
  "name": "$DEST_TBL",
  "spec": {
    "source_table_full_name": "$SRC_TBL",
    "primary_key_columns": $PKS,
    "scheduling_policy": "TRIGGERED",
  },
  "new_pipeline_spec": {
    "storage_catalog": "$ST_CATALOG",
    "storage_schema": "$ST_SCHEMA",
  }
}
EOF

Creare una tabella sincronizzata in un catalogo standard di Unity Catalog.

export CATALOG_NAME=<Standard catalog>
export DATABASE_INSTANCE=<database instance>
export POSTGRES_DATABASE=$CATALOG_NAME
export SRC_TBL="source_catalog.source_schema.source_table"
export DEST_TBL="$CATALOG_NAME.some_schema.sync_table"
export PKS='["id"]'
export ST_CATALOG="storage_catalog"
export ST_SCHEMA="storage_schema"

curl -X POST https://$WORKSPACE/api/2.0/database/synced_tables \
-H "Content-Type: text/json" \
-H "Authorization: Bearer ${DATABRICKS_TOKEN}" \
--data-binary @- << EOF
{
  "name": "$DEST_TBL",
  "database_instance_name": "$DATABASE_INSTANCE",
  "logical_database_name": "$POSTGRES_DATABASE",
  "spec": {
    "source_table_full_name": "$SRC_TBL",
    "primary_key_columns": $PKS,
    "scheduling_policy": "TRIGGERED",
  },
  "new_pipeline_spec": {
    "storage_catalog": "$ST_CATALOG",
    "storage_schema": "$ST_SCHEMA",
  }
}
EOF

Controllare lo stato di una tabella sincronizzata.

export SYNCEDTABLE='pg_db.silver.sbtest1_online'

curl --request GET \
  "https://e2-dogfood.staging.cloud.databricks.com/api/2.0/database/synced_tables/$SYNCEDTABLE" \
  --header "Authorization: Bearer dapi..."

Modalità di sincronizzazione illustrate

È possibile creare una tabella sincronizzata con una delle modalità di sincronizzazione seguenti, che determinano la modalità di sincronizzazione dei dati dall'origine alla tabella sincronizzata in Postgres:

Pianificare o attivare le sincronizzazioni successive

Alla creazione, lo snapshot iniziale viene eseguito automaticamente. Per le modalità snapshot e attivate , le sincronizzazioni successive devono essere attivate in modo esplicito. La modalità continua si gestisce automaticamente.

Attività di sincronizzazione della tabella del database nella pipeline

L'attività della pipeline di sincronizzazione delle tabelle del database nei Processi Lakeflow esegue la pipeline di una tabella sincronizzata come passaggio nel flusso di lavoro. Configurare l'attività con un trigger di aggiornamento tabella o una pianificazione.

Trigger per gli aggiornamenti delle tabelle di origine

Genera il compito quando viene aggiornata la tabella del Catalogo Unity di origine. Con la modalità attivata , vengono applicate in modo incrementale solo le nuove modifiche, offrendo aggiornamenti quasi in tempo reale senza il costo sempre attivo della modalità continua.

1. Nella barra laterale fare clic su Flussi di lavoro.
2. Fare clic su Crea processo o aprire un processo esistente.
3. Nella scheda Attività fare clic su + Aggiungi un altro tipo di attività.
4. In Inserimento e trasformazione selezionare Pipeline di sincronizzazione tabelle di database.
5. Nel campo Pipeline selezionare la pipeline associata alla tabella sincronizzata.
6. In Pianificazioni e trigger fare clic su Aggiungi trigger.
7. Selezionare Aggiornamento tabella come tipo di trigger.
8. In Tabelle selezionare la tabella del catalogo Unity di origine da monitorare.
9. Fare clic su Salva.

Generare attività in base a una pianificazione

Esegue la sincronizzazione a cadenza fissa. Ideale per la modalità snapshot , in cui un aggiornamento completo notturno o settimanale è in genere il modello più efficiente.

1. Seguire i passaggi da 1 a 5 precedenti per aggiungere un'attività della pipeline di sincronizzazione tabelle di database a un processo.
2. In Pianificazioni e trigger fare clic su Aggiungi trigger.
3. Scegli Pianificato come tipo di trigger.
4. Impostare la pianificazione cron e il fuso orario, quindi fare clic su Salva.

Uso dell'SDK

Attivare un'esecuzione di sincronizzazione a livello di codice, ad esempio alla fine di un notebook o di una pipeline upstream:

from databricks.sdk import WorkspaceClient

w = WorkspaceClient()

# Get the pipeline ID from the synced table
table = w.database.get_synced_database_table(name="catalog.schema.synced_table")
pipeline_id = table.data_synchronization_status.pipeline_id

# Trigger a sync run
w.pipelines.start_update(pipeline_id=pipeline_id)

Operazioni supportate

Azure Databricks consiglia di eseguire solo le operazioni seguenti in Postgres per le tabelle sincronizzate per evitare sovrascrizioni accidentali o incoerenze dei dati:

• Query di sola lettura
• Creazione di indici
• Eliminazione della tabella (per liberare spazio dopo la rimozione della tabella sincronizzata dal catalogo unity)

anche se è possibile modificare le tabelle sincronizzate in Postgres in altri modi, interferisce con la pipeline di sincronizzazione.

Eliminare una tabella sincronizzata

Per eliminare una tabella sincronizzata, è necessario eliminarla da Unity Catalog e quindi eliminare la tabella nell'istanza del database. L'eliminazione della tabella sincronizzata da Unity Catalog annulla la registrazione della tabella e arresta eventuali aggiornamenti dei dati. Tuttavia, la tabella rimane nel database Postgres sottostante. Per liberare spazio nell'istanza del database, connettersi all'istanza e usare il DROP TABLE comando .

INTERFACCIA UTENTE

1. Fare clic su Catalogo nella barra laterale dell'area di lavoro.
2. Trovare e selezionare la tabella sincronizzata da eliminare.
3. Fare clic >Elimina.
4. Connetti all'istanza con psql, l'editor SQL o da un notebook.
5. Eliminare la tabella usando PostgreSQL.

   DROP TABLE synced_table_database.synced_table_schema.synced_table
   

PYTHON SDK

from databricks.sdk import WorkspaceClient

# Initialize the Workspace client
w = WorkspaceClient()

# Delete a synced table from UC
synced_table_name = "catalog.schema.synced_table"
w.database.delete_synced_database_table(name=synced_table_name)
print(f"Deleted synced table from UC: {synced_table_name}")

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

CLI

# Delete a synced table from UC
databricks database delete-synced-database-table "catalog.schema.synced_table"

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

curva

# Delete a synced table from UC
curl -X DELETE --header "Authorization: Bearer ${DATABRICKS_TOKEN}" \
  https://$WORKSPACE/api/2.0/database/synced_tables/$SYNCED_TABLE_NAME

# To free up space in your database instance, you need to connect to the
# instance and drop the table using PostgreSQL:
#
# DROP TABLE synced_table_database.synced_table_schema.synced_table;

Proprietà e autorizzazioni

Se si crea un nuovo database, uno schema o una tabella Postgres, la proprietà postgres viene impostata come segue:

• La proprietà viene assegnata all'utente che crea il database, lo schema o la tabella, se l'account di accesso di Azure Databricks esiste come ruolo in Postgres. Per aggiungere un ruolo di identità di Azure Databricks in Postgres, vedere Gestire i ruoli postgreSQL.
• In caso contrario, la proprietà viene assegnata al proprietario dell'oggetto padre in Postgres (in genere ).databricks_superuser

Gestire l'accesso alle tabelle sincronizzate

Dopo aver creato una tabella sincronizzata, è databricks_superuser possibile READ sincronizzare una tabella da Postgres. Il databricks_superuser ha pg_read_all_data, che consente a questo ruolo di leggere da tutte le tabelle. Ha anche il pg_write_all_data privilegio, che consente a questo ruolo di scrivere su tutte le tabelle. Ciò significa che un databricks_superuser oggetto può anche scrivere in una tabella sincronizzata in Postgres. Lakebase supporta questo comportamento di scrittura nel caso in cui sia necessario apportare modifiche urgenti nella tabella di destinazione. Tuttavia, Databricks consiglia di apportare correzioni nella tabella di origine.

• Il databricks_superuser può anche concedere questi privilegi ad altri utenti:

  GRANT USAGE ON SCHEMA synced_table_schema TO user;
  

  GRANT SELECT ON synced_table_name TO user;
  

• databricks_superuser può revocare questi privilegi:

  REVOKE USAGE ON SCHEMA synced_table_schema FROM user;
  

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

Gestire le operazioni delle tabelle sincronizzate

databricks_superuser Può gestire gli utenti autorizzati a eseguire operazioni specifiche su una tabella sincronizzata. Le operazioni supportate per le tabelle sincronizzate sono:

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

Tutte le altre operazioni DDL vengono negate per le tabelle sincronizzate.

Per concedere questi privilegi ad altri utenti, è databricks_superuser necessario innanzitutto creare un'estensione in databricks_auth:

CREATE EXTENSION IF NOT EXISTS databricks_auth;

databricks_superuser Può quindi aggiungere un utente per gestire una tabella sincronizzata:

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

Può databricks_superuser rimuovere un utente dalla gestione di una tabella sincronizzata:

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

Il sistema databricks_superuser può visualizzare tutti i responsabili:

SELECT * FROM databricks_synced_table_managers;

Mappatura del tipo di dati

Questa tabella di mappatura dei tipi definisce come ogni tipo di dati nella tabella del catalogo Unity di origine viene mappato alla tabella di sincronizzazione di destinazione in Postgres.

Gestire caratteri non validi

Alcuni caratteri, ad esempio il byte Null (0x00), sono consentiti nelle STRINGcolonne , ARRAY, MAPo STRUCT nelle tabelle Delta, ma non sono supportati in Postgres TEXT o JSONB colonne. Di conseguenza, la sincronizzazione di tali dati da Delta a Postgres può causare errori di inserimento.

org.postgresql.util.PSQLException: ERROR: invalid byte sequence for encoding "UTF8": 0x00

org.postgresql.util.PSQLException: ERROR: unsupported Unicode escape sequence DETAIL: \u0000 cannot be converted to text.

• Il primo errore si verifica quando compare un byte nullo in una colonna stringa di primo livello, che esegue il mapping diretto a Postgres TEXT.
• Il secondo errore si verifica quando un byte Null viene visualizzato in una stringa annidata all'interno di un tipo complesso (STRUCT, ARRAYo MAP), che Azure Databricks serializza come JSONB. Durante la serializzazione, tutte le stringhe vengono convertite in Postgres TEXT, dove \u0000 non è consentito.

Soluzione:

È possibile risolvere questo problema in uno dei modi seguenti:

• Purificare i campi stringa

  Rimuovere o sostituire caratteri non supportati da tutti i campi stringa, inclusi quelli all'interno di tipi complessi, prima di eseguire la sincronizzazione con Postgres.

  Per rimuovere byte null da una colonna di primo livello STRING, usare la funzione REPLACE.

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

• Converti in binario (solo per STRING le colonne)

  Se è necessario conservare il contenuto dei byte non elaborati, convertire le colonne interessate STRING in BINARY.

Limitazioni e considerazioni

Tabelle di origine supportate

A seconda della modalità di sincronizzazione di una tabella sincronizzata, sono supportati diversi tipi di tabelle di origine:

• Per la modalità snapshot, la tabella di origine deve supportare SELECT *. Ad esempio: tabelle Delta, tabelle Iceberg, viste, viste materializzate e altri tipi simili.

• Per le modalità di sincronizzazione attivate o continue, la tabella di origine deve avere abilitato ancheil feed di dati delle modifiche .

Limitazioni dei nomi e degli identificatori

• Caratteri consentiti: I nomi di database, schema e tabella Postgres per le tabelle sincronizzate possono contenere solo caratteri alfanumerici e caratteri di sottolineatura ([A-Za-z0-9_]+). I trattini (-) e altri caratteri speciali non sono supportati.
• Identificatori di colonna e tabella: Evitare di usare lettere maiuscole o caratteri speciali nei nomi di colonna o tabella nella tabella del catalogo Unity di origine. Se mantenuto, è necessario citare questi identificatori quando si fa riferimento a tali identificatori in Postgres.

Prestazioni e sincronizzazione

• Velocità di sincronizzazione: La sincronizzazione dei dati in tabelle sincronizzate può risultare più lenta rispetto alla scrittura dei dati direttamente nell'istanza del database con un client PostgreSQL nativo a causa di un'ulteriore elaborazione. La modalità di sincronizzazione continua aggiorna i dati dalla tabella gestita di Unity Catalog alla tabella sincronizzata con un intervallo minimo di 15 secondi.
• Utilizzo della connessione: Ogni sincronizzazione di tabelle può usare fino a 16 connessioni all'istanza del database, che vengono conteggiati per il limite di connessione dell'istanza.
• Idempotenza API: Le API delle tabelle sincronizzate sono idempotenti e potrebbe essere necessario ripetere l'operazione quando si verificano errori per garantire operazioni tempestive.
• Modifiche dello schema: Per le tabelle sincronizzate in Triggered modalità o Continuous , nella tabella sincronizzata vengono riflesse solo le modifiche dello schema aggiuntive (ad esempio, l'aggiunta di una nuova colonna) dalle tabelle del catalogo Unity.
• Chiavi duplicate: Se due righe hanno la stessa chiave primaria nella tabella di origine, la pipeline di sincronizzazione ha esito negativo a meno che non si configuri la deduplicazione usando una chiave timeseries. Tuttavia, l'uso di una chiave serie temporale comporta una penalizzazione delle prestazioni.
• Frequenza di aggiornamento: Per Lakebase Provisioned, la pipeline di sincronizzazione supporta scritture continue e scritture attivate a circa 1200 righe al secondo per unità di capacità (CU) e scritture snapshot fino a 15.000 righe al secondo per CU.

Capacità e limiti

• Limiti delle tabelle:
  • Limite di 20 tabelle sincronizzate per tabella di origine.
  • Ogni sincronizzazione delle tabelle può usare fino a 16 connessioni di database.
• Limiti di dimensioni e aggiornamento completo:
  • Se si aggiorna completamente una tabella sincronizzata, la versione precedente in Postgres non viene eliminata fino a quando non viene sincronizzata la nuova tabella. Entrambe le versioni vengono conteggiate temporaneamente al limite di dimensioni del database logico durante l'aggiornamento.
  • Le singole tabelle sincronizzate non hanno un limite, ma il limite totale delle dimensioni dei dati logici per tutte le tabelle nell'istanza è di 2 TB. Tuttavia, se sono necessari aggiornamenti anziché la ricreazione completa della tabella, Databricks consiglia di non superare 1 TB.
  • Se le dimensioni non compresse del formato di riga della tabella del catalogo Unity superano il limite di dimensioni dell'istanza del database (2 TB), la sincronizzazione non riesce. È necessario eliminare la tabella sincronizzata in Postgres prima di scrivere ulteriormente nell'istanza.

Integrazione del catalogo

• Duplicazione del catalogo: La creazione di una tabella sincronizzata in un catalogo standard destinata a un database Postgres registrato anche come catalogo di database separato fa sì che la tabella sincronizzata venga visualizzata in Unity Catalog sia con i cataloghi standard che con i cataloghi di database.