Archivos de ingesta como tipo de archivo

Importante

Esta característica se encuentra en su versión beta. Los administradores del área de trabajo pueden controlar el acceso a esta característica desde la página Vistas previas . Consulte Administrar versiones preliminares de Azure Databricks.

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.

Nota:

FILE Las columnas no tienen un orden definido. No puedes usar una FILE columna como columna de partición, columna de agrupamiento o clave de orden Z. Para obtener más información, consulte Límites.

Modos de almacenamiento

Una FILE referencia puede almacenarse en uno de dos modos:

FILE EXTERNAL referencias que ya existen en un volumen del Catálogo de Unity. Databricks no permite almacenar FILE EXTERNAL referencias para archivos almacenados fuera de volúmenes.
FILE MANAGED almacena copias de archivos en almacenamiento gestionado por el Catálogo de Unity. Los archivos procedentes de fuentes externas a volúmenes, como SharePoint, Google Drive o SFTP, deben ser ingeridos y almacenados como FILE MANAGED.

Uso `list_files` para descubrir archivos

Utiliza la list_files función de tabla para descubrir 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 valorada en tabla.

Ingesta de archivos como referencias de archivo

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

Ingesta archivos de volumen como ARCHIVO EXTERNO

Para ingerir archivos que ya existen en un volumen del Catálogo de Unity, utilice una CREATE TABLE AS SELECT instrucción (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/');

Ingirir archivos fuente 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()

Utiliza pipelines para ingirir nuevos archivos de forma incremental

Para ingerir nuevos archivos a medida que llegan, utiliza una tabla de streaming en una tubería Lakeflow que lee la fuente con 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 encender pipelines declarativos.

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

Configura el canal de la tubería en PREVIEW. Ingerir FILE referencias en una tubería requiere el PREVIEW canal.

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.

Warning

Databricks recomienda que primero atiendas los datos de cambio en una tabla gestionada, como en el siguiente ejemplo, y luego apliques AUTO CDC a esa tabla. Aplicar AUTO CDC directamente a STREAM read_files(..., readChangeFeed => true) relee la fuente de cambios de fuente para cada flujo descendente, lo que podría aumentar los costes de procesamiento.

Ingire el cambio en dos pasos. El siguiente ejemplo ingiere la fuente de cambios de SharePoint y luego la aplica a una tabla de streaming destino como SCD tipo 1:

Escribe los datos de cambio en una tabla de streaming con archivos gestionados, como en el siguiente código. Activa readChangeFeed => trueread_files para devolver el feed de cambios, que incluye las columnas _file_id, _sequence, y _is_deleted metadatos.

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

Ú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 un volumen como ARCHIVO EXTERNO

Para escribir los archivos en un volumen del Catálogo de Unity como archivos externos, pasa a 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()

Escribe datos binarios en almacenamiento gestionado como FILE MANAGED

Para almacenar los archivos como archivos gestionados, llama 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()

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

Comentarios

¿Le ha resultado útil esta página?

Last updated on 2026-08-03

Archivos de ingesta como tipo de archivo

Modos de almacenamiento

Uso list_files para descubrir archivos

Ingesta de archivos como referencias de archivo

Ingesta archivos de volumen como ARCHIVO EXTERNO

Ingirir archivos fuente externos como FILE MANAGED

SQL

Python

Scala

Utiliza pipelines para ingirir nuevos archivos de forma incremental

SQL

Python

Aplica actualizaciones y eliminaciones con AUTO CDC

SQL

Python

SQL

Python

Convertir datos binarios en línea en referencias de ARCHIVO

Escribe datos binarios en un volumen como ARCHIVO EXTERNO

SQL

Python

Scala

Escribe datos binarios en almacenamiento gestionado como FILE MANAGED

SQL

Python

Scala

Pasos siguientes

Comentarios

Recursos adicionales

Uso `list_files` para descubrir archivos