Uso de Auto Loader con Unity Catalog

Auto Loader puede ingerir datos de manera segura desde ubicaciones externas que están configuradas con Unity Catalog. Para más información sobre cómo conectar de forma segura el almacenamiento con el catálogo de Unity, consulte Conexión al almacenamiento de objetos en la nube mediante el catálogo de Unity. Auto Loader se basa en Structured Streaming para el procesamiento incremental; para obtener recomendaciones y limitaciones, consulte Uso de Unity Catalog con Structured Streaming.

Especificar ubicaciones para los recursos de Auto Loader para el catálogo de Unity

El modelo de seguridad de Unity Catalog supone que todas las ubicaciones de almacenamiento a las que se hace referencia en una carga de trabajo se administrarán mediante Unity Catalog. Databricks recomienda almacenar siempre la información de evolución de puntos de control y esquemas en ubicaciones de almacenamiento que administra Unity Catalog. Unity Catalog no permite anidar archivos de puntos de control ni archivos de inferencia y evolución de esquemas en el directorio de la tabla.

Ingesta de datos desde el almacenamiento en la nube mediante el catálogo de Unity

En los ejemplos siguientes se supone que el usuario en ejecución tiene READ FILES permisos en la ubicación externa, los privilegios de propietario en las tablas de destino y las siguientes configuraciones y concesiones.

Uso del cargador automático para cargar en una tabla administrada del catálogo de Unity

En los ejemplos siguientes se muestra cómo usar Auto Loader para ingerir datos en una tabla administrada del catálogo de 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'
);

Cuando usas read_files en una sentencia CREATE STREAMING TABLE dentro de las canalizaciones declarativas de Lakeflow para Spark, los puntos de control y las ubicaciones del esquema se gestionan automáticamente.

Utilice Auto Loader para cargar datos en una tabla externa de Unity Catalog

Para mantener los datos en una ubicación de almacenamiento específica, use una tabla externa de Catálogo de Unity en lugar de una tabla administrada. Por ejemplo, use una tabla externa para compartir datos con clientes que no son de Databricks o para registrar datos existentes. Con las tablas externas, defines la ruta de almacenamiento. Vea Trabajar con tablas externas.

Para usar Auto Loader con una tabla externa de Unity Catalog, primero registre la tabla con CREATE TABLE ... LOCATION y, a continuación, haga streaming en ella por nombre. La ubicación de la tabla debe estar dentro de una ubicación externa donde tenga CREATE EXTERNAL TABLE permisos. La ubicación del punto de control también debe estar en una ubicación externa administrada por el catálogo de Unity. Use una ruta de acceso independiente de los datos de la tabla.

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