Importar archivos como tipo FILE

El FILE tipo almacena y consulta referencias a archivos no estructurados (documentos, imágenes y audio) en tablas. Esta página muestra cómo descubrir archivos, ingerirlos como FILE referencias e incorporar nuevos archivos de forma incremental a medida que llegan.

Para la referencia sobre el FILE tipo, véase FILE tipo. Para una visión general de los enfoques para la ingesta de datos no estructurados, véase TIPO de archivo y datos no estructurados.

Modos de almacenamiento

Una FILE referencia puede almacenarse en uno de dos modos:

• FILE MANAGED almacena copias de archivos en almacenamiento administrado de Unity Catalog: los permisos se administran a través de la tabla y eliminar filas hace que los archivos referenciados puedan optar a la recolección de elementos no utilizados, de modo que la tabla y sus archivos permanecen sincronizados. Los archivos procedentes de orígenes fuera de volúmenes, como SharePoint, Google Drive o SFTP, deben ingerirse y almacenarse como FILE MANAGED.
• FILE EXTERNAL hace referencia a archivos que ya existen en un volumen de Unity Catalog. Databricks no permite almacenar FILE EXTERNAL referencias para archivos almacenados fuera de volúmenes.

Azure Databricks recomienda FILE MANAGED para cargas de trabajo que se benefician de permisos a nivel de archivo y cumplimiento integrado. Para una comparación entre gobernanza y comportamiento del ciclo de vida, véase Tipo de archivo y datos no estructurados.

Uso list_files para descubrir archivos

Utiliza la list_files función con valores de tabla función con valores de tabla para detectar los archivos disponibles en una ruta. Devuelve una fila por archivo con sus path, size, modification_time, y una FILE referencia:

SELECT * FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

Para descubrir archivos en una fuente que requiera una conexión con Unity Catalog, como SharePoint, Google Drive o SFTP, añade el connection parámetro:

SELECT * FROM list_files('https://example.sharepoint.com/sites/my-site/', connection => 'my_sharepoint_connection');

list_files descubre archivos recursivamente por defecto. Para saber más, véase list_files función con valores de tabla.

Ingesta de archivos como referencias de archivo

Selecciona un método de ingestión según dónde almacenes tus archivos. Para ingir archivos de una fuente externa, cópialos al almacenamiento gestionado como FILE MANAGED. Para referenciar archivos que ya están en un volumen del Catálogo de Unity sin copiarlos, use FILE EXTERNAL.

Incorporar archivos de origen externos como FILE MANAGED

Para generar FILE referencias para archivos en una fuente como SharePoint, Google Drive o SFTP, ingiere primero los archivos y guárdalos como FILE MANAGED. FILE EXTERNAL No se soporta para archivos almacenados fuera de volúmenes.

El siguiente ejemplo ingiere archivos de SharePoint en una FILE MANAGED tabla:

SQL

CREATE TABLE managed_documents (
  file_name STRING,
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents
  SELECT _metadata.file_name, *
  FROM read_files(
    'https://example.sharepoint.com/sites/my-site/',
    connection => 'my_sharepoint_connection',
    format => 'file');

Python

(spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append())

Scala

spark.read.format("file")
  .option("databricks.connection", "my_sharepoint_connection")
  .load("https://example.sharepoint.com/sites/my-site/")
  .selectExpr("_metadata.file_name", "*")
  .writeTo("managed_documents").append()

Ingesta archivos de volumen como ARCHIVO EXTERNO

Para cargar archivos que ya existen en un volumen de Unity Catalog, utilice una instrucción CREATE TABLE AS SELECT (CTAS) con list_files. Esto crea una tabla con una FILE EXTERNAL columna que hace referencia a cada archivo en su lugar, sin copiar su contenido. El siguiente ejemplo crea una documents tabla con el nombre del archivo, los metadatos y una FILE referencia para cada archivo:

CREATE TABLE documents AS
  SELECT _metadata.file_name, *
  FROM list_files('/Volumes/my_catalog/my_schema/raw_files/');

Utiliza pipelines para ingirir nuevos archivos de forma incremental

Para ingerir nuevos archivos a medida que llegan, utiliza una tabla de streaming en una canalización de Lakeflow que lee el origen mediante STREAM read_files(..., format => 'file'). Cada actualización de pipeline procesa solo los archivos añadidos después de la última actualización. Ver read_files y Spark Declarative Pipelines on Lakeflow.

Para transmitir archivos de forma incremental desde una fuente como Google Drive:

1. Configura el canal de la tubería en PREVIEW. La incorporación de referencias FILE en una canalización requiere el canal PREVIEW.

2. Definamos una tabla de streaming que lea la fuente con STREAM read_files(..., format => 'file'), como en el siguiente código:

   SQL

   CREATE STREAMING TABLE streaming_documents (
     path STRING,
     size BIGINT,
     modification_time TIMESTAMP,
     file FILE MANAGED
   )
   TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
   AS SELECT *
     FROM STREAM read_files(
       'https://drive.google.com/drive/folders/my-folder-id',
       connection => 'my_gdrive_connection',
       format => 'file');
   

   Python

   from pyspark import pipelines as dp
   
   @dp.table(
     name="streaming_documents",
     schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   def streaming_documents():
     return (
       spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "file")
         .option("databricks.connection", "my_gdrive_connection")
         .load("https://drive.google.com/drive/folders/my-folder-id")
     )
   

Aplica actualizaciones y eliminaciones con AUTO CDC

Una ingesta de streaming añade nuevos archivos pero no captura actualizaciones ni eliminaciones de la fuente. Para aplicar esos cambios, lee el feed de cambios de código fuente con AUTO CDC.

Ingiera la fuente de cambios en dos pasos. El siguiente ejemplo incorpora el flujo de cambios de SharePoint y luego lo aplica a una tabla de streaming de destino como SCD de tipo 1:

1. Escribe los datos de cambio en una tabla de streaming con archivos gestionados, como en el siguiente código. Configura readChangeFeed => true en read_files para devolver la fuente de cambios, que incluye las columnas de metadatos _file_id, _sequence y _is_deleted.

   SQL

   CREATE OR REFRESH STREAMING TABLE documents_changes (
     _file_id STRING,
     _sequence BIGINT,
     _is_deleted BOOLEAN,
     path STRING,
     size BIGINT,
     modification_time TIMESTAMP,
     file FILE MANAGED
   )
   TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
   AS SELECT *
     FROM STREAM read_files(
       'https://example.sharepoint.com/sites/my-site/',
       connection => 'my_sharepoint_connection',
       format => 'file',
       readChangeFeed => true);
   

   Python

   from pyspark import pipelines as dp
   
   @dp.table(
     name="documents_changes",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   def documents_changes():
     return (
       spark.readStream.format("cloudFiles")
         .option("cloudFiles.format", "file")
         .option("databricks.connection", "my_sharepoint_connection")
         .option("cloudFiles.readChangeFeed", "true")
         .load("https://example.sharepoint.com/sites/my-site/")
     )
   

2. Úsalo AUTO CDC para aplicar los cambios de esa tabla a una tabla de streaming objetivo, como en el siguiente código. Úsalo _file_id como clave, _sequence como columna de secuencia y _is_deleted para identificar eliminaciones.

   SQL

   CREATE OR REFRESH STREAMING TABLE documents
     TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');
   
   CREATE FLOW documents_cdc AS AUTO CDC INTO
     documents
   FROM STREAM documents_changes
     KEYS (_file_id)
     APPLY AS DELETE WHEN _is_deleted = true
     SEQUENCE BY _sequence
     COLUMNS * EXCEPT (_is_deleted, _sequence)
     STORED AS SCD TYPE 1;
   

   Python

   from pyspark import pipelines as dp
   from pyspark.sql.functions import col, expr
   
   dp.create_streaming_table(
     name="documents",
     table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
   )
   
   dp.create_auto_cdc_flow(
     target = "documents",
     source = "documents_changes",
     keys = ["_file_id"],
     sequence_by = col("_sequence"),
     apply_as_deletes = expr("_is_deleted = true"),
     except_column_list = ["_is_deleted", "_sequence"],
     stored_as_scd_type = 1
   )
   

Convertir datos binarios en línea en referencias de ARCHIVO

Si una tabla ya almacena el contenido del archivo como datos binarios en línea, usa create_file la función para escribir esos datos y producir una FILE referencia.

Los siguientes ejemplos utilizan una tabla generada por el usuario, raw_documents, con una name columna y una content columna que contienen los datos binarios.

Escribe datos binarios en almacenamiento gestionado como FILE MANAGED

Para almacenar los archivos como archivos gestionados, llama a create_file solo con el contenido binario. Cuando omites destination_path, Unity Catalog sube el contenido a la ubicación de almacenamiento gestionado:

SQL

CREATE TABLE managed_documents (name STRING, file FILE MANAGED) USING DELTA
  TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/');

INSERT INTO managed_documents (name, file)
  SELECT name, create_file(content => content)
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr("name", "create_file(content => content) AS file")
  .writeTo("managed_documents").append()

Escribe datos binarios en un volumen como ARCHIVO EXTERNO

Para escribir los archivos en un volumen del Catálogo Unity como archivos externos, pasa un destination_path a create_file, como en el siguiente código:

SQL

CREATE TABLE documents (name STRING, file FILE EXTERNAL) USING DELTA;

INSERT INTO documents (name, file)
  SELECT
    name,
    create_file(
      content => content,
      destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name
    )
  FROM raw_documents;

Python

(spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append())

Scala

spark.read.table("raw_documents")
  .selectExpr(
    "name",
    "create_file(content => content, destination_path => '/Volumes/my_catalog/my_schema/my_volume/' || name) AS file")
  .writeTo("documents").append()

Pasos siguientes

• FILE tipo
• Tipo de archivo y datos no estructurados
• Tutorial: Construye una cadena de procesamiento de archivos con el tipo ARCHIVO
• Más información sobre Auto Loader. Consulte ¿Qué es Auto Loader?.