Hantera lakehouse-data med synkroniserade tabeller (Lakebase Provisioned)

Med synkroniserade tabeller kan du hantera lakehouse-data via Lakebase Provisioned Postgres-instanser. Unity Catalog-tabeller synkroniseras med Postgres så att program kan fråga lakehouse-data direkt med låg svarstid. Den här processen kallas ofta omvänd ETL. lakehouse är optimerat för analys och berikning, medan Lakebase är utformat för operativa arbetsbelastningar som kräver snabba sökningar och transaktionskonsekvens.

Vad är synkroniserade tabeller?

Med synkroniserade tabeller kan du hantera analysdata från Unity Catalog via Lakebase Provisioned Postgres, vilket gör dem tillgängliga för program som behöver frågor med låg svarstid och fullständiga ACID-transaktioner. De överbryggar klyftan mellan analytisk lagring och driftsystem genom att hålla dina data redo att användas i realtidsprogram.

Synkroniseringspipelines använder hanterade Lakeflow Spark-deklarativa pipelines för att kontinuerligt uppdatera både den synkroniserade tabellen Unity Catalog och Postgres-tabellen med ändringar från källtabellen. När du har skapat dem kan synkroniserade tabeller frågas direkt med Postgres-verktyg.

De viktigaste egenskaperna för synkroniserade tabeller är följande:

• Tabellen behandlas som skrivskyddad i Postgres för att upprätthålla dataintegriteten med källan
• Synkroniseras automatiskt med hanterade Spark-deklarativa pipelines för Lakeflow
• Frågebara via Standard PostgreSQL-gränssnitt
• Hanteras via Unity Catalog för styrning och livscykelhantering

Innan du börjar

• Du har en Unity Catalog-tabell i valfri katalog.
• Du har CAN USE behörighet för databasinstansen.

Skapa en synkroniserad tabell

Användargränssnitt (UI)

Om du vill synkronisera en Unity Catalog-tabell till Postgres gör du följande:

1. Klicka på Katalog i arbetsytans sidofält.

2. Leta upp och välj den Unity Catalog-tabell som du vill skapa en synkroniserad tabell på.

3. Klicka på Skapa>synkroniserad tabell.

4. Välj din katalog, ditt schema och ange ett tabellnamn för den nya synkroniserade tabellen.

   • Synkroniserade tabeller kan också skapas i Standard-kataloger med ytterligare konfiguration. Välj standardkatalogen, ett schema och ange ett tabellnamn för den nyligen skapade synkroniserade tabellen.

5. Välj en databasinstans och ange namnet på Postgres-databasen där den synkroniserade tabellen ska skapas. Postgres-databasfältet är som standard den valda målkatalogen. Om en Postgres-databas inte finns under det här namnet skapar Azure Databricks en ny.

6. Välj en primärnyckel. En primärnyckel krävs eftersom den ger effektiv åtkomst till rader för läsningar, uppdateringar och borttagningar.

   Viktigt!

   Kolumner i primärnyckeln kan inte vara null i den synkroniserade tabellen. Därför undantas rader med null-värden i primärnyckelkolumner från synkroniseringen.

7. Om två rader har samma primärnyckel i källtabellen väljer du en timeserienyckel för att konfigurera deduplicering. När en timeserienyckel har angetts innehåller de synkroniserade tabellerna endast raderna med det senaste tidsserienyckelvärdet för varje primärnyckel.

8. Välj synkroniseringsläge från Ögonblicksbild, Utlöst och Kontinuerlig. Mer information om varje synkroniseringsläge finns i Beskrivning av synkroniseringslägen.

9. Välj om du vill skapa den här synkroniserade tabellen från en ny eller befintlig pipeline.

   • Om du skapar en ny pipeline och använder en hanterad katalog ska du välja lagringsplats för staging-tabellen. Om du använder en standardkatalog lagras staging-tabellen automatiskt i katalogen.
   • Om du använder en befintlig pipeline kontrollerar du att det nya synkroniseringsläget matchar pipelineläget.

10. (Valfritt) Välj en serverlös användningsprincip. Information om hur du skapar en serverlös användningsprincip finns i Attributanvändning med serverlösa användningsprinciper. På så sätt kan du tillskriva faktureringsanvändning till specifika serverlösa användningsprinciper.

    • För synkroniserade tabeller är den fakturerbara entiteten den underliggande Lakeflow Spark Deklarativ pipeline. Om du vill ändra budgetprincipen ändrar du det underliggande pipelineobjektet. Se Konfigurera en serverlös pipeline.

11. När Status för synkroniserad tabell är Online loggar du in på din databasinstans och frågar den nyligen skapade tabellen. Utför frågor på din tabell med SQL-redigeraren, externa verktyg eller anteckningsböcker.

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}")

Kommandoradsgränssnitt (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"

lockig

Skapa en synkroniserad tabell i en databaskatalog.

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

Skapa en synkroniserad tabell i en standardkatalog i 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

Kontrollera statusen för en synkroniserad tabell.

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..."

Beskrivning av synkroniseringslägen

En synkroniserad tabell kan skapas med något av följande synkroniseringslägen, som avgör hur data synkroniseras från källan till den synkroniserade tabellen i Postgres:

Schemalägga eller utlösa efterföljande synkroniseringar

Den första "snapshoten" körs automatiskt vid skapandet. För Snapshot-läge och Trigger-läge måste efterföljande synkroniseringar utlösas explicit. Kontinuerligt läge är självhanterande.

Pipelineuppgift för synkronisering av databastabell

Pipelineaktiviteten Databastabellsynkronisering i Lakeflow-jobb kör en synkroniserad tabells pipeline som ett arbetsflödessteg. Konfigurera jobbet med en tabelluppdateringsutlösare eller ett schema.

Utlösare för källtabelluppdateringar

Jobbet utlöses när den Unity Catalog-källtabellen uppdateras. I Triggad läge tillämpas endast nya ändringar stegvis, vilket ger uppdatering i nära realtid utan kostnaden för Kontinuerligt läge.

1. I sidofältet klickar du på Arbetsflöden.
2. Klicka på Skapa jobb eller öppna ett befintligt jobb.
3. På fliken Uppgifter klickar du på + Lägg till en annan aktivitetstyp.
4. Under Inmatning och transformering väljer du Pipeline för databastabellsynkronisering.
5. I fältet Pipeline väljer du den pipeline som är associerad med den synkroniserade tabellen.
6. Under Scheman och utlösare klickar du på Lägg till utlösare.
7. Välj Tabelluppdatering som utlösartyp.
8. Under Tabeller väljer du den unity catalog-källtabell som ska övervakas.
9. Klicka på Spara.

Aktivera på ett schema

Kör synkroniseringen med en fast takt. Passar bra för läget Ögonblicksbild , där en hel uppdatering varje natt eller vecka vanligtvis är det mest effektiva mönstret.

1. Följ steg 1–5 ovan för att lägga till en Database Table Sync pipeline-uppgift i ett jobb.
2. Under Scheman och utlösare klickar du på Lägg till utlösare.
3. Välj Schemalagd som utlösartyp.
4. Ange cron-schemat och tidszonen och klicka sedan på Spara.

Användning av SDK

Utlös en synkroniseringskörning programmatiskt, till exempel i slutet av en överordnad notebook-fil eller pipeline:

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)

Åtgärder som stöds

Azure Databricks rekommenderar att du endast utför följande åtgärder i Postgres för synkroniserade tabeller för att förhindra oavsiktliga överskrivningar eller datainkonsekvenser:

• Skrivskyddade förfrågningar
• Skapa index
• Ta bort tabellen (för att frigöra utrymme när du har tagit bort den synkroniserade tabellen från Unity Catalog)

Även om det är möjligt att ändra synkroniserade tabeller i Postgres på andra sätt, stör det synkroniseringspipelinen.

Ta bort en synkroniserad tabell

Om du vill ta bort en synkroniserad tabell måste du ta bort den från Unity Catalog och sedan släppa tabellen i databasinstansen. Om du tar bort den synkroniserade tabellen från Unity Catalog avregistreras tabellen och alla datauppdateringar stoppas. Tabellen finns dock kvar i den underliggande Postgres-databasen. Om du vill frigöra utrymme i databasinstansen ansluter du till instansen och använder DROP TABLE kommandot .

Användargränssnitt (UI)

1. Klicka på Katalog i arbetsytans sidofält.
2. Leta upp och välj den synkroniserade tabell som du vill ta bort.
3. Klicka på >Ta bort.
4. Anslut till instansen med psql, SQL-redigeraren eller från en notebook-fil.
5. Ta bort tabellen med 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;

Kommandoradsgränssnitt (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;

lockig

# 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;

Ägarskap och behörigheter

Om du skapar en ny Postgres-databas, ett schema eller en ny tabell anges Postgres-ägarskapet på följande sätt:

• Ägarskap tilldelas användaren som skapar databasen, schemat eller tabellen, om deras Azure Databricks inloggning finns som en roll i Postgres. Om du vill lägga till en Azure Databricks identitetsroll i Postgres läser du Hantera PostgreSQL-roller.
• I annat fall tilldelas ägarskapet till ägaren av det överordnade objektet i Postgres (vanligtvis databricks_superuser).

Hantera synkroniserad tabellåtkomst

När en synkroniserad tabell har skapats databricks_superuser kan READ en synkroniserad tabell från Postgres. databricks_superuser har pg_read_all_data, som tillåter den här rollen att läsa från alla tabeller. Den har också behörigheten pg_write_all_data , vilket gör att den här rollen kan skriva till alla tabeller. Det innebär att en databricks_superuser även kan skriva till en synkroniserad tabell i Postgres. Lakebase stöder det här skrivbeteendet om du behöver göra brådskande ändringar i måltabellen. Databricks rekommenderar dock att du gör korrigeringar i källtabellen i stället.

• databricks_superuser Kan också bevilja dessa behörigheter till andra användare:

  GRANT USAGE ON SCHEMA synced_table_schema TO user;
  

  GRANT SELECT ON synced_table_name TO user;
  

• databricks_superuser Kan återkalla dessa behörigheter:

  REVOKE USAGE ON SCHEMA synced_table_schema FROM user;
  

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

Hantera synkroniserade tabellåtgärder

databricks_superuser Kan hantera vilka användare som har behörighet att utföra specifika åtgärder i en synkroniserad tabell. De åtgärder som stöds för synkroniserade tabeller är:

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

Alla andra DDL-åtgärder nekas för synkroniserade tabeller.

Om du vill bevilja dessa behörigheter till ytterligare användare databricks_superuser måste du först skapa ett tillägg på databricks_auth:

CREATE EXTENSION IF NOT EXISTS databricks_auth;

databricks_superuser Sedan kan lägga till en användare för att hantera en synkroniserad tabell:

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

databricks_superuser Kan ta bort en användare från att hantera en synkroniserad tabell:

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

databricks_superuser Kan visa alla chefer:

SELECT * FROM databricks_synced_table_managers;

Datatypkartläggning

Den här typen av mappningstabell definierar hur varje datatyp i unity catalog-källtabellen mappas till målsynkroniseringstabellen i Postgres:

Hantera ogiltiga tecken

Vissa tecken, till exempel null byte (0x00), tillåts i STRING, ARRAY, MAPeller STRUCT kolumner i Delta-tabeller men stöds inte i Postgres TEXT eller JSONB kolumner. Därför kan synkronisering av sådana data från Delta till Postgres leda till infogningsfel och andra felmeddelanden.

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.

• Det första felet uppstår när en nullbyte visas i en strängkolumn på toppnivå, som mappar direkt till Postgres TEXT.
• Det andra felet uppstår när en null-byte visas i en sträng kapslad i en komplex typ (STRUCT, ARRAY eller MAP), som Azure Databricks serialiserar som JSONB. Under serialiseringen kastas alla strängar till Postgres TEXT, där \u0000 inte tillåts.

Så här löser du problemet:

Du kan åtgärda det här problemet på något av följande sätt:

• Rensa strängfält

  Ta bort eller ersätt tecken som inte stöds från alla strängfält, inklusive de i komplexa typer, innan du synkroniserar till Postgres.

  Om du vill ta bort null-byte från en kolumn på den översta nivån STRING använder du REPLACE funktionen:

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

• Konvertera till binär (endast för STRING kolumner)

  Om du behöver bevara råbyteinnehållet konverterar du de berörda STRING kolumnerna till BINARY.

Begränsningar och överväganden

Källtabeller som stöds

Beroende på synkroniseringsläget för en synkroniserad tabell stöds olika typer av källtabeller:

• För ögonblicksbildläget måste källtabellen ha stöd för SELECT *. Exempel är Delta-tabeller, Isbergstabeller, vyer, materialiserade vyer och andra liknande typer.

• För lägen för utlöst eller kontinuerlig synkronisering måste källtabellen också ha aktiverat ändringsdataflöde .

Begränsningar för namngivning och identifierare

• Tillåtna tecken: Postgres-databas-, schema- och tabellnamn för synkroniserade tabeller får endast innehålla alfanumeriska tecken och understreck ([A-Za-z0-9_]+). Bindestreck (-) och andra specialtecken stöds inte.
• Kolumn- och tabellidentifierare: Undvik att använda versaler eller specialtecken i kolumn- eller tabellnamn i unity catalog-källtabellen. Om de bevaras måste du citera dessa identifierare när du refererar till dem i Postgres.

Prestanda och synkronisering

• Synkroniseringshastighet: Att synkronisera data i synkroniserade tabeller kan vara långsammare än att skriva data direkt till databasinstansen med en intern PostgreSQL-klient på grund av ytterligare bearbetning. Läget för kontinuerlig synkronisering uppdaterar data från den hanterade tabellen Unity Catalog till den synkroniserade tabellen med ett minsta intervall på 15 sekunder.
• Anslutningsanvändning: Varje tabellsynkronisering kan använda upp till 16 anslutningar till databasinstansen, vilket räknas mot instansens anslutningsgräns.
• API-idempotens: API:er för synkroniserade tabeller är idempotenter och kan behöva göras om om fel uppstår för att säkerställa åtgärder i tid.
• Schemaändringar: För synkroniserade tabeller i Triggered eller Continuous läge återspeglas endast tilläggsschemaändringar (t.ex. tillägg av en ny kolumn) från Unity Catalog-tabellerna i den synkroniserade tabellen.
• Duplicerade nycklar: Om två rader har samma primärnyckel i källtabellen misslyckas synkroniseringspipelinen om du inte konfigurerar deduplicering med hjälp av en tidsserienyckel. Men att använda en tidsserienyckel medför en prestationsnedsättning.
• Uppdateringsfrekvens: För Lakebase Provisioned stöder synkroniseringspipelinen kontinuerliga och utlösta skrivningar på cirka 1 200 rader per sekund per kapacitetsenhet (CU) och ögonblicksbildsskrivningar på upp till 15 000 rader per sekund per CU.

Kapacitet och gränser

• Tabellgränser:
  • Gräns på 20 synkroniserade tabeller per källtabell.
  • Varje tabellsynkronisering kan använda upp till 16 databasanslutningar.
• Storleksgränser och fullständig uppdatering:
  • Om du uppdaterar en synkroniserad tabell tas inte den gamla versionen i Postgres bort förrän den nya tabellen har synkroniserats. Båda versionerna räknas tillfälligt mot storleksgränsen för den logiska databasen under uppdateringen.
  • Enskilda synkroniserade tabeller har ingen gräns, men den totala logiska datastorleksgränsen för alla tabeller i instansen är 2 TB. Men om du behöver uppdateringar i stället för fullständig table recreation rekommenderar Databricks att du inte överskrider 1 TB.
  • Om den okomprimerade radformatstorleken för Unity Catalog-tabellen överskrider storleksgränsen för databasinstansen (2 TB) misslyckas synkroniseringen. Du måste släppa den synkroniserade tabellen i Postgres innan du skriver vidare till instansen.

Katalogintegrering

• Katalogduplicering: Om du skapar en synkroniserad tabell i en standardkatalog som riktar sig mot en Postgres-databas som också är registrerad som en separat databaskatalog visas den synkroniserade tabellen i Unity Catalog under både standard- och databaskatalogerna.