Utilisation d’Auto Loader avec Unity Catalog

Auto Loader peut ingérer des données en toute sécurité depuis des emplacements externes configurés avec Unity Catalog. Pour en savoir plus sur la connexion sécurisée du stockage avec Unity Catalog, consultez Se connecter au stockage d'objets cloud à l'aide de Unity Catalog. Auto Loader repose sur Structured Streaming pour le traitement incrémentiel ; pour les recommandations et les limitations, consultez Utilisation de Unity Catalog avec Structured Streaming.

remarque

Dans Databricks Runtime 11.3 LTS et versions ultérieures, vous pouvez utiliser Auto Loader avec des modes d’accès standard ou dédié (anciennement modes d’accès partagé et mono-utilisateur).

Le mode de listage de répertoires est pris en charge par default.

Spécifiez les emplacements des ressources Auto Loader pour Unity Catalog

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 des schémas dans des emplacements de stockage gérés par Unity Catalog. Unity Catalog ne vous permet pas d'imbriquer les fichiers de points de contrôle, d'inférence de schéma et d'évolution sous le répertoire de la table.

Ingérer des données du stockage cloud à l’aide d’Unity Catalog

Les exemples suivants supposent que l'utilisateur qui exécute la commande dispose des autorisations READ FILES sur l'emplacement externe, des privilèges de propriétaire sur les tables cibles, ainsi que des configurations et des autorisations suivantes.

Emplacement de stockage.	Accorder
s3://autoloader-source/json-data	READ FILES
s3://dev-bucket	READ FILES, WRITE FILES, CREATE TABLE

Emplacement de stockage.	Accorder
s3://autoloader-source/json-data	READ FILES
s3://dev-bucket	READ FILES, WRITE FILES, CREATE TABLE

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

Les exemples suivants montrent comment utiliser Auto Loader pour ingérer des données dans une table gérée par Unity Catalog.

Python

Python

checkpoint_path = "s3://dev-bucket/_checkpoint/dev_table"

(spark.readStream
  .format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", checkpoint_path)
  .load("s3://autoloader-source/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))

SQL

SQL

CREATE OR REFRESH STREAMING TABLE dev_catalog.dev_database.dev_table
AS SELECT * FROM STREAM read_files(
  's3://autoloader-source/json-data',
  format => 'json'
);

Lorsque vous utilisez read_files dans une instruction CREATE STREAMING TABLE dans les LakeFlow Pipelines, les emplacements des checkpoints et des schémas sont gérés automatiquement.

Utilisez Auto Loader pour charger dans une table externe Unity Catalog

Pour conserver les 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 enregistrer des données existantes. Avec les tables externes, vous définissez le chemin de stockage. Consultez Travailler avec des tables externes.

Pour utiliser Auto Loader avec une table externe Unity Catalog, commencez par enregistrer la table avec CREATE TABLE ... LOCATION, puis effectuez le Stream vers celle-ci par son nom. L’emplacement de la table doit se trouver dans un emplacement externe où 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 Unity Catalog. Utilisez un chemin séparé des données de la table.

Python

checkpoint_path = "s3://dev-bucket/_checkpoint/dev_table"
table_path = "s3://dev-bucket/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("s3://autoloader-source/json-data")
  .writeStream
  .option("checkpointLocation", checkpoint_path)
  .trigger(availableNow=True)
  .toTable("dev_catalog.dev_database.dev_table"))