Synchroniser les données des tables du catalogue Unity vers une instance de base de données

Cette page explique comment créer et gérer une table synchronisée. Une table synchronisée est une table Postgres en lecture seule de catalogue Unity qui synchronise automatiquement les données d’une table Unity Catalog avec votre instance de base de données Lakebase. La synchronisation d’une table de catalogue Unity dans Postgres permet des requêtes de lecture à faible latence et prend en charge les jointures au moment des requêtes avec d’autres tables Postgres.

La synchronisation est gérée par les pipelines déclaratifs Spark Lakeflow. Un pipeline managé met à jour en permanence la table Postgres avec des modifications de la table source. Après la création, les tables synchronisées peuvent être interrogées directement à l’aide d’outils Postgres.

Les principales caractéristiques des tables synchronisées sont les suivantes :

• Mode lecture seule dans Postgres pour maintenir l’intégrité des données par rapport à la source.
• Synchronisé automatiquement à l’aide de pipelines déclaratifs Spark gérés Lakeflow
• Interrogeable via des interfaces PostgreSQL standard
• Géré via le catalogue Unity pour la gouvernance et la gestion du cycle de vie

Avant de commencer

• Vous disposez d’une table de catalogue Unity dans n’importe quel catalogue.
• Vous disposez des CAN USE autorisations sur l’instance de base de données.

Créer une table synchronisée

IU

Pour synchroniser une table de catalogue Unity dans Postgres, procédez comme suit :

1. Cliquez sur Catalogue dans la barre latérale de l’espace de travail.

2. Recherchez et sélectionnez la table catalogue Unity sur laquelle vous souhaitez créer une table synchronisée.

3. Cliquez sur Créer une>table synchronisée.

4. Sélectionnez votre catalogue, votre schéma et entrez un nom de table pour la nouvelle table synchronisée.

   • Les tables synchronisées peuvent également être créées dans des catalogues Standard, avec une configuration supplémentaire. Sélectionnez votre catalogue Standard, un schéma et entrez un nom de table pour la table synchronisée nouvellement créée.

5. Sélectionnez une instance de base de données et entrez le nom de la base de données Postgres dans laquelle créer la table synchronisée. Le champ de base de données Postgres est défini par défaut sur le catalogue cible actuellement sélectionné. Si une base de données Postgres n’existe pas sous ce nom, Azure Databricks en crée une nouvelle.

6. Sélectionnez une clé primaire. Une clé primaire est requise , car elle permet un accès efficace aux lignes pour les lectures, les mises à jour et les suppressions.

   Important

   Les colonnes de la clé primaire ne peuvent pas être nulles dans la table synchronisée. Par conséquent, les lignes avec des valeurs Null dans les colonnes de clé primaire sont exclues de la synchronisation.

7. Si deux lignes ont la même clé primaire dans la table source, sélectionnez une clé Timeseries pour configurer la déduplication. Lorsqu’une clé Timeseries est spécifiée, les tables synchronisées contiennent uniquement les lignes avec la dernière valeur de clé timeseries pour chaque clé primaire.

8. Sélectionnez le mode de synchronisation parmi Instantané, Déclenché et Continu. Pour plus d’informations sur chaque mode de synchronisation, consultez les modes de synchronisation expliqués.

9. Choisissez si vous souhaitez créer cette table synchronisée à partir d’un pipeline nouveau ou existant.

   • Si vous créez un pipeline et utilisez un catalogue managé, choisissez l’emplacement de stockage de la table intermédiaire. Si vous utilisez un catalogue standard, la table intermédiaire est automatiquement stockée dans le catalogue.
   • Si vous utilisez un pipeline existant, vérifiez que le nouveau mode de synchronisation correspond au mode de pipeline.

10. (Facultatif) Sélectionnez une stratégie de budget serverless. Pour créer une stratégie de budget serverless, consultez l’utilisation des attributs avec des stratégies de budget serverless. Cela vous permet d’attribuer l’utilisation de la facturation à des stratégies d’utilisation spécifiques.

    • Pour les tables synchronisées, l’entité facturable est le pipeline sous-jacent des pipelines déclaratifs de Lakeflow Spark. Pour modifier la stratégie budgétaire, modifiez l’objet de pipeline sous-jacent. Consultez Configurer un pipeline serverless.

11. Une fois que l’état de la table synchronisée est En ligne, connectez-vous à votre instance de base de données et interrogez la table nouvellement créée. Interrogez votre table à l’aide de l’éditeur SQL, des outils externes ou des notebooks.

Kit de développement logiciel (SDK) Python

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

Interface de ligne de commande (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"

friser

Créez une table synchronisée dans un catalogue de bases de données.

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

Créez une table synchronisée dans un catalogue de catalogues Unity standard.

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

Vérifiez l’état d’une table synchronisée.

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

Modes de synchronisation expliqués

Une table synchronisée peut être créée avec l’un des modes de synchronisation suivants, qui déterminent comment les données sont synchronisées de la source vers la table synchronisée dans Postgres :

Opérations prises en charge

Databricks recommande d’effectuer uniquement les opérations suivantes dans Postgres pour les tables synchronisées afin d’éviter les remplacements accidentels ou les incohérences de données :

• Requêtes en lecture seule
• Création d’index
• Suppression de la table (pour libérer de l’espace après avoir supprimé la table synchronisée du catalogue Unity)

Bien que vous puissiez modifier cette table de différentes façons, elle interfère avec le pipeline de synchronisation.

Supprimer une table synchronisée

Pour supprimer une table synchronisée, vous devez la supprimer du catalogue Unity, puis supprimer la table dans l’instance de base de données. La suppression de la table synchronisée à partir du catalogue Unity annule l’inscription de la table et arrête toutes les actualisations de données. Toutefois, la table reste dans la base de données Postgres sous-jacente. Pour libérer de l’espace dans votre instance de base de données, connectez-vous à l’instance et utilisez la DROP TABLE commande.

IU

1. Cliquez sur Catalogue dans la barre latérale de l’espace de travail.
2. Recherchez et sélectionnez la table synchronisée à supprimer.
3. Cliquez sur >Supprimez.
4. Connectez-vous à l’instance avec psql, l’éditeur SQL ou à partir d’un notebook.
5. Supprimez la table à l’aide de PostgreSQL.

   DROP TABLE synced_table_database.synced_table_schema.synced_table
   

Kit de développement logiciel (SDK) Python

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;

Interface de ligne de commande (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;

friser

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

Propriété et autorisations

Si vous créez une base de données, un schéma ou une table Postgres, la propriété Postgres est définie comme suit :

• La propriété est affectée à l’utilisateur qui crée la base de données, le schéma ou la table, si sa connexion Azure Databricks existe en tant que rôle dans Postgres. Pour ajouter un rôle d’identité Azure Databricks dans Postgres, consultez Gérer les rôles Postgres.
• Sinon, la propriété est affectée au propriétaire de l’objet parent dans Postgres (généralement le databricks_superuser).

Gérer l’accès aux tables synchronisées

Une fois qu’une table synchronisée est créée, la databricks_superuser peut READ être une table synchronisée à partir de Postgres. databricks_superuserA pg_read_all_data , qui permet à ce rôle de lire à partir de toutes les tables. Il possède également le privilège pg_write_all_data, ce qui permet à ce rôle d’écrire dans toutes les tables. Cela signifie qu’un databricks_superuser peut également écrire dans une table synchronisée dans Postgres. Lakebase prend en charge ce comportement d’écriture si vous devez apporter des modifications urgentes dans votre table cible. Toutefois, Databricks recommande d’apporter des correctifs dans votre table source à la place.

• databricks_superuser peut également accorder ces privilèges à d'autres utilisateurs :

  GRANT USAGE ON SCHEMA synced_table_schema TO user;
  

  GRANT SELECT ON synced_table_name TO user;
  

• databricks_superuser peut révoquer ces privilèges :

  REVOKE USAGE ON SCHEMA synced_table_schema FROM user;
  

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

Gérer les opérations de table synchronisée

Les databricks_superuser peuvent gérer quels utilisateurs sont autorisés à effectuer des opérations spécifiques sur une table synchronisée. Les opérations prises en charge pour les tables synchronisées sont les suivantes :

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

Toutes les autres opérations DDL sont refusées pour les tables synchronisées.

Pour accorder ces privilèges à des utilisateurs supplémentaires, databricks_superuser doit d’abord créer une extension sur databricks_auth:

CREATE EXTENSION IF NOT EXISTS databricks_auth;

Ensuite, databricks_superuser peut ajouter un utilisateur pour gérer une table synchronisée :

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

Le databricks_superuser peut supprimer un utilisateur de la gestion d’une table synchronisée.

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

databricks_superuser peut afficher tous les gestionnaires :

SELECT * FROM databricks_synced_table_managers;

Mappage de types de données

Cette table de mappage de type définit la façon dont chaque type de données de la table catalogue Unity source est mappé à la table de synchronisation de destination dans Postgres :

Gérer les caractères non valides

Certains caractères, tels que l’octet Null (0x00), sont autorisés dans les colonnes STRING, ARRAY, MAP ou STRUCT des tables Delta, mais ne sont pas pris en charge dans les colonnes TEXT ou JSONB de Postgres. Par conséquent, la synchronisation de ces données de Delta vers Postgres peut entraîner des échecs d’insertion avec des erreurs :

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.

• La première erreur se produit lorsqu’un octet Null apparaît dans une colonne de chaîne de niveau supérieur, qui est mappée directement à Postgres TEXT.
• La deuxième erreur se produit lorsqu’un octet Null apparaît dans une chaîne imbriquée à l’intérieur d’un type complexe (STRUCT, ARRAY ou MAP), que sérialise Azure Databricks en tant que JSONB. Pendant la sérialisation, toutes les chaînes sont converties en Postgres TEXT, où \u0000 est interdit.

Résolution du problème :

Vous pouvez résoudre ce problème de l’une des manières suivantes :

• Nettoyer les champs de chaîne

  Supprimez ou remplacez les caractères non pris en charge de tous les champs de chaîne, y compris ceux à l’intérieur de types complexes, avant la synchronisation avec Postgres.

  Pour supprimer des octets null d’une colonne de niveau STRING supérieur, utilisez la REPLACE fonction :

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

• Convertir en binaire (pour les STRING colonnes uniquement)

  Si la conservation du contenu d’octet brut est nécessaire, convertissez les colonnes affectées de STRING en BINARY.

Limitations et considérations

Tables sources prises en charge

Selon le mode de synchronisation d’une table synchronisée, différents types de tables sources sont pris en charge :

• Pour le mode Instantané, la table source doit prendre en charge SELECT *. Les exemples incluent des tables Delta, des tables icebergs, des vues matérialisées et d’autres types similaires.

• Pour les modes de synchronisation déclenchée ou continue, la table source doit également avoir activé le flux de données de modification .

Limitations de nommage et d’identificateur

• Caractères autorisés : Les noms de base de données, de schéma et de table Postgres pour les tables synchronisées peuvent contenir uniquement des caractères alphanumériques et des traits de soulignement ([A-Za-z0-9_]+). Les traits d’union (-) et d’autres caractères spéciaux ne sont pas pris en charge.
• Identificateurs de colonne et de table : Évitez d’utiliser des lettres majuscules ou des caractères spéciaux dans les noms de colonnes ou de tables dans la table catalogue Unity source. Si vous le conservez, vous devez citer ces identificateurs lors de leur référencement dans Postgres.

Performances et synchronisation

• Vitesse de synchronisation : La synchronisation des données dans des tables synchronisées peut être plus lente que l’écriture de données directement dans l’instance de base de données avec un client PostgreSQL natif en raison d’un traitement supplémentaire. Le mode de synchronisation continue actualise les données de la table managée du catalogue Unity vers la table synchronisée à un intervalle minimal de 15 secondes.
• Utilisation de la connexion : Chaque synchronisation de tables peut utiliser jusqu’à 16 connexions à l’instance de base de données, ce qui compte pour la limite de connexion de l’instance.
• Idempotence de l'API : Les API de table synchronisées sont idempotentes et peuvent nécessiter une nouvelle tentative en cas d'erreurs pour garantir la réalisation opportune des opérations.
• Modifications de schéma : Pour les tables synchronisées en mode Triggered ou Continuous, seules les modifications de schéma additives (par exemple, l’ajout d’une nouvelle colonne) à partir des tables du catalogue Unity sont reflétées sur la table synchronisée.
• Clés en double : Si deux lignes ont la même clé primaire dans la table source, le pipeline de synchronisation échoue, sauf si vous configurez la déduplication à l’aide d’une clé Timeseries. Toutefois, l’utilisation d’une clé de série chronologique entraîne une pénalité de performance.
• Taux de mise à jour : Le pipeline de synchronisation prend en charge les écritures continues à environ 1 200 lignes par seconde par unité de capacité (CU) et les écritures en bloc jusqu’à 15 000 lignes par seconde par cu.

Capacité et limites

• Limites de table :
  • Limite de 20 tables synchronisées par table source.
  • Chaque synchronisation de tables peut utiliser jusqu’à 16 connexions de base de données.
• Limites de taille et actualisation complète :
  • Si vous actualisez entièrement une table synchronisée, l’ancienne version dans Postgres n’est pas supprimée tant que la nouvelle table n’est pas synchronisée. Les deux versions comptent temporairement pour la limite de taille de la base de données logique lors de l’actualisation.
  • Les tables synchronisées individuelles n’ont pas de limite, mais la limite de taille totale des données logiques sur toutes les tables de l’instance est de 2 To. Toutefois, si vous avez besoin d’actualisations au lieu de la reconstruction complète de la table, Databricks recommande de ne pas dépasser 1 téraoctet.
  • Si la taille de format de ligne non compressée de la table catalogue Unity dépasse la limite de taille d’instance de base de données (2 To), la synchronisation échoue. Vous devez supprimer la table synchronisée dans Postgres avant d’écrire plus loin dans l’instance.

Intégration du catalogue

• Duplication de catalogue : La création d’une table synchronisée dans un catalogue standard ciblant une base de données Postgres qui est également inscrite en tant que catalogue de bases de données distinct entraîne l’affichage de la table synchronisée dans le catalogue Unity sous les catalogues standard et de base de données.