Utilisation du chargeur automatique avec Unity Catalog

Le chargeur automatique peut ingérer en toute sécurité des données à partir d’emplacements externes configurés avec Unity Catalog. Pour en savoir plus sur la connexion sécurisée du stockage avec le catalogue Unity, consultez Se connecter au stockage d’objets cloud à l’aide du catalogue Unity. Le chargeur automatique s’appuie sur Structured Streaming pour le traitement incrémentiel ; pour obtenir des recommandations et des limitations, consultez Utilisation d’Unity Catalog avec Structured Streaming.

Spécifier des emplacements pour les ressources d'Auto Loader pour le catalogue Unity

Le modèle de sécurité Unity Catalog suppose que tous les emplacements de stockage référencés dans une charge de travail seront gérés par Unity Catalog. Databricks recommande de toujours stocker les informations de point de contrôle et d’évolution de schéma dans les emplacements de stockage gérés par Unity Catalog. Unity Catalog ne vous permet pas d’imbriquer des fichiers d’inférence et d’évolution de point de contrôle ou de schéma sous le répertoire de table.

Ingérer des données à partir du stockage cloud à l’aide du catalogue Unity

Les exemples suivants supposent que l’utilisateur en cours d’exécution dispose d’autorisations READ FILES sur l’emplacement externe, les privilèges de propriétaire sur les tables cibles et les configurations et octrois suivants.

Utiliser Auto Loader pour charger dans une table gérée par Unity Catalog

Les exemples suivants montrent comment utiliser l'Auto Loader pour ingérer des données dans une table managée du catalogue Unity.

Python

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))

SQL

CREATE OR REFRESH STREAMING TABLE dev_catalog.dev_database.dev_table
AS SELECT * FROM STREAM read_files(
  'abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data',
  format => 'json'
);

Lorsque vous utilisez read_files dans une instruction CREATE STREAMING TABLE au sein des pipelines Lakeflow, les emplacements du point de contrôle et du schéma sont gérés automatiquement.

Utiliser le chargeur automatique pour charger dans une table externe du catalogue Unity

Pour conserver des données dans un emplacement de stockage spécifique, utilisez une table externe Unity Catalog au lieu d’une table gérée. Par exemple, utilisez une table externe pour partager des données avec des clients non Databricks ou pour inscrire des données existantes. Avec les tables externes, vous définissez le chemin d’accès au stockage. Consultez Utiliser des tables externes.

Pour utiliser le chargeur automatique avec une table externe Unity Catalog, enregistrez d’abord la table avec CREATE TABLE ... LOCATION, puis alimentez-la en flux par son nom. L’emplacement de la table doit se trouver dans un emplacement externe pour lequel vous disposez des autorisations CREATE EXTERNAL TABLE. L’emplacement du point de contrôle doit également se trouver dans un emplacement externe géré par le catalogue Unity. Utilisez un chemin d’accès distinct des données de la table.

checkpoint_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/_checkpoint/dev_table"
table_path = "abfss://dev-bucket@<storage-account>.dfs.core.windows.net/external/dev_table"

# One-time: register the external table in UC.
spark.sql(f"""
  CREATE TABLE IF NOT EXISTS dev_catalog.dev_database.dev_table
  USING DELTA
  LOCATION '{table_path}'
""")

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("abfss://autoloader-source@<storage-account>.dfs.core.windows.net/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))