Trabajar con datos no estructurados en volúmenes

En esta página se muestra cómo almacenar, consultar y procesar archivos de datos no estructurados mediante volúmenes de Catálogo de Unity. Aprenderá a cargar archivos, consultar metadatos, procesar archivos con funciones de IA, aplicar el control de acceso y compartir volúmenes con otras organizaciones. Siempre que sea posible, se han incluido instrucciones para trabajar en este tutorial mediante la interfaz de usuario del Explorador de catálogos. Si no se muestra ninguna opción del Explorador de catálogos , use los comandos SQL o Python proporcionados.

Para obtener información general completa sobre las funcionalidades de volumen y los casos de uso, consulte ¿Qué son los volúmenes del catálogo de Unity?.

Requisitos

• Un área de trabajo de Azure Databricks con Unity Catalog habilitado.
• CREATE CATALOG privilegio en el metastore. Consulte Creación de catálogos. Si no puede crear un catálogo, pida al administrador acceso o use un catálogo existente en el que tenga el CREATE SCHEMA privilegio.
• Databricks Runtime 14.3 LTS y versiones posteriores.
• Para las funciones de IA: un área de trabajo en una región admitida.
• Para OpenSharing: se requieren los privilegios CREATE SHARE y CREATE RECIPIENT sobre el metastore. Consulte Uso compartido de datos y recursos de inteligencia artificial de forma segura.

Paso 1: Crear un volumen

Cree un catálogo, un esquema y un volumen para almacenar los archivos. Para obtener instrucciones detalladas sobre la administración de volúmenes, consulte Creación y administración de volúmenes del catálogo de Unity.

Paso 1.1: Crear un catálogo y un esquema

SQL

-- Create a catalog
CREATE CATALOG IF NOT EXISTS unstructured_data_lab;
USE CATALOG unstructured_data_lab;

-- Create a schema
CREATE SCHEMA IF NOT EXISTS raw;
USE SCHEMA raw;

Pitón

spark.sql("CREATE CATALOG IF NOT EXISTS unstructured_data_lab")
spark.sql("USE CATALOG unstructured_data_lab")
spark.sql("CREATE SCHEMA IF NOT EXISTS raw")
spark.sql("USE SCHEMA raw")

Explorador de catálogos

1. Haga clic en Catálogo en la barra lateral.
2. Haga clic en Crear>un catálogo.
3. Escriba unstructured_data_lab como nombre del catálogo.
4. Haga clic en Crear.
5. Haga clic en Ver catálogo.

En la página del catálogo:

1. Haga clic en Crear esquema.
2. Escriba raw como nombre del esquema.
3. Haga clic en Crear.

Paso 1.2: Creación de un volumen administrado

SQL

CREATE VOLUME IF NOT EXISTS files_volume
COMMENT 'Volume for storing unstructured data files';

Pitón

spark.sql("""
    CREATE VOLUME IF NOT EXISTS files_volume
    COMMENT 'Volume for storing unstructured data files'
""")

Explorador de catálogos

En la página de esquema:

1. Haga clic en Crear>Volumen.
2. Escriba files_volume como nombre del volumen.
3. Compruebe que el volumen administrado está seleccionado.
4. Haga clic en Crear.

Paso 2: Cargar archivos

Cargue archivos en su volumen. Para obtener ejemplos completos de administración de archivos, consulte Trabajar con archivos en volúmenes del catálogo de Unity.

Paso 2.1: Cargar archivos

Puede usar ejemplos de databricks-datasets para este tutorial o cargar sus propios archivos mediante la interfaz de usuario del Explorador de catálogos.

Pitón

# Upload a single image file
dbutils.fs.cp(
    "dbfs:/databricks-datasets/flower_photos/roses/10090824183_d02c613f10_m.jpg",
    "/Volumes/unstructured_data_lab/raw/files_volume/rose.jpg"
)

# Upload a single PDF file
dbutils.fs.cp(
    "dbfs:/databricks-datasets/COVID/CORD-19/2020-03-13/COVID.DATA.LIC.AGMT.pdf",
    "/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf"
)

# Upload a directory
local_dir = "dbfs:/databricks-datasets/samples/data/mllib"
volume_path = "/Volumes/unstructured_data_lab/raw/files_volume/sample_files"

for file_info in dbutils.fs.ls(local_dir):
    source = file_info.path
    dest = f"{volume_path}/{file_info.name}"
    dbutils.fs.cp(source, dest, recurse=True)
    print(f"Uploaded: {file_info.name}")

Explorador de catálogos

El código de Python de la pestaña Python sube dos archivos (un JPG y un PDF) y un directorio que incluye archivos .txt y .csv. Para cargar archivos mediante el Explorador de catálogos:

1. En la página del volumen, haga clic en Cargar en este volumen.
2. En el cuadro de diálogo Cargar archivos, en Archivos, haga clic en Examinar o arrastre y suelte archivos en el área de colocación.
3. En Volumen de destino, compruebe que el volumen que creó en el paso anterior está seleccionado.

Paso 2.2: Comprobar la carga

SQL

LIST '/Volumes/unstructured_data_lab/raw/files_volume/';

Pitón

files = dbutils.fs.ls("/Volumes/unstructured_data_lab/raw/files_volume/")
for f in files:
    print(f"{f.name}\t{f.size} bytes")

Explorador de catálogos

Cuando se cargan los archivos, aparecen en la página del volumen. Haga clic en un nombre de archivo para ver una vista previa o haga clic en un directorio para ver archivos individuales.

%fs ls /Volumes/unstructured_data_lab/raw/files_volume/

Paso 3: Consulta de metadatos de archivo

Consulte la información del archivo para comprender lo que hay en el volumen. Para obtener más patrones de consulta, consulte Enumeración y consulta de archivos en volúmenes con SQL.

Paso 3.1: Mostrar metadatos de archivo

SQL

SELECT
  path,
  _metadata.file_name,
  _metadata.file_size,
  _metadata.file_modification_time
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile'
);

Pitón

df = (
    spark.read
    .format("binaryFile")
    .option("recursiveFileLookup", "true")
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")
)

df.select("path", "modificationTime", "length").show(truncate=False)

Explorador de catálogos

La página de volumen del Explorador de catálogos muestra el nombre de cada archivo (incluida la extensión), el tamaño y la fecha de última modificación .

Paso 4: Consultar y procesar archivos

Use las funciones de Inteligencia artificial de Azure Databricks para extraer contenido de documentos y analizar imágenes. Para obtener información general completa sobre las funcionalidades de las funciones de inteligencia artificial, consulte Enriquecimiento de datos mediante AI Functions.

Paso 4.1: Análisis de documentos

SQL

SELECT
  path AS file_path,
  ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.pdf'
);

Pitón

result_df = spark.sql("""
    SELECT
      path AS file_path,
      ai_parse_document(content, map('version', '2.0')) AS parsed_content
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.pdf'
    )
""")
display(result_df)

%pip install PyPDF2==3.0.1

from pyspark.sql.functions import udf
from pyspark.sql.types import StringType
from PyPDF2 import PdfReader
import io

@udf(returnType=StringType())
def extract_pdf_text(content):
    if content is None:
        return None
    try:
        reader = PdfReader(io.BytesIO(content))
        return "\n".join(page.extract_text() or "" for page in reader.pages)
    except Exception as e:
        return f"Error: {str(e)}"

df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")

result_df = df.withColumn("text_content", extract_pdf_text("content"))
display(result_df.select("path", "text_content"))

Paso 4.2: Análisis de imágenes

SQL

SELECT
  path,
  ai_query(
    'databricks-llama-4-maverick',
    'Describe this image in one sentence:',
    files => content
  ) AS description
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.{jpg,jpeg,png}'
)
WHERE _metadata.file_size < 5000000;

Pitón

result_df = spark.sql("""
    SELECT
      path,
      ai_query(
        'databricks-llama-4-maverick',
        'Describe this image in one sentence:',
        files => content
      ) AS description
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.{jpg,jpeg,png}'
    )
    WHERE _metadata.file_size < 5000000
""")
display(result_df)

%pip install pillow==10.4.0

from pyspark.sql.functions import udf
from pyspark.sql.types import StructType, StructField, IntegerType, StringType
from PIL import Image
import io

image_schema = StructType([
    StructField("width", IntegerType()),
    StructField("height", IntegerType()),
    StructField("format", StringType())
])

@udf(returnType=image_schema)
def get_image_info(content):
    if content is None:
        return None
    try:
        img = Image.open(io.BytesIO(content))
        return {"width": img.width, "height": img.height, "format": img.format}
    except:
        return None

df = spark.read.format("binaryFile") \
    .option("pathGlobFilter", "*.{jpg,jpeg,png}") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")

result_df = df.withColumn("image_info", get_image_info("content"))
display(result_df.select("path", "image_info.*"))

Paso 4.3: Filtrar y analizar por nombre de archivo

En este ejemplo se filtran los archivos de imagen con la subcadena "rose" en su nombre de archivo.

SQL

SELECT
  path AS file_path,
  ai_query(
    'databricks-llama-4-maverick',
    'Describe this image in one sentence:',
    files => content
  ) AS description
FROM read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.{jpg,jpeg,png}'
)
WHERE _metadata.file_name ILIKE '%rose%';

Pitón

result_df = spark.sql("""
    SELECT
      path AS file_path,
      ai_query(
        'databricks-llama-4-maverick',
        'Describe this image in one sentence:',
        files => content
      ) AS description
    FROM read_files(
      '/Volumes/unstructured_data_lab/raw/files_volume/',
      format => 'binaryFile',
      fileNamePattern => '*.{jpg,jpeg,png}'
    )
    WHERE _metadata.file_name ILIKE '%rose%'
""")
display(result_df)

Paso 4.4: Combinar archivos con tablas estructuradas

En este ejemplo se usan números de fila para emparejar archivos con viajes de taxi con fines de demostración. En producción, únase a claves empresariales significativas.

SQL

-- This example demonstrates joining file metadata with structured data
-- by pairing files with taxi trips using row numbers
WITH files_with_row AS (
  SELECT
    path,
    SPLIT(path, '/')[SIZE(SPLIT(path, '/')) - 1] AS file_name,
    length,
    ROW_NUMBER() OVER (ORDER BY path) AS file_row
  FROM read_files(
    '/Volumes/unstructured_data_lab/raw/files_volume/',
    format => 'binaryFile'
  )
),
trips_with_row AS (
  SELECT
    tpep_pickup_datetime,
    pickup_zip,
    dropoff_zip,
    fare_amount,
    ROW_NUMBER() OVER (ORDER BY tpep_pickup_datetime) AS trip_row
  FROM samples.nyctaxi.trips
  WHERE pickup_zip IS NOT NULL
  LIMIT 5
)
SELECT
  f.path,
  f.file_name,
  f.length,
  t.pickup_zip,
  t.dropoff_zip,
  t.fare_amount,
  t.tpep_pickup_datetime
FROM files_with_row f
INNER JOIN trips_with_row t ON f.file_row = t.trip_row;

Pitón

from pyspark.sql.functions import col, row_number, element_at, split
from pyspark.sql.window import Window

# Read files and add row numbers
files_df = spark.read.format("binaryFile") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/") \
    .withColumn("file_name", element_at(split(col("path"), "/"), -1))

files_with_row = files_df.alias("files") \
    .withColumn("file_row", row_number().over(Window.orderBy("path")))

# Get trips and add row numbers
trips_df = spark.table("samples.nyctaxi.trips") \
    .filter(col("pickup_zip").isNotNull()) \
    .limit(5)

trips_with_row = trips_df.alias("trips") \
    .withColumn("trip_row", row_number().over(Window.orderBy("tpep_pickup_datetime")))

# Join on row numbers
result_df = files_with_row \
    .join(trips_with_row, col("file_row") == col("trip_row"), "inner") \
    .select(
        "files.path",
        "files.file_name",
        "files.length",
        "trips.pickup_zip",
        "trips.dropoff_zip",
        "trips.fare_amount",
        "trips.tpep_pickup_datetime"
    )

display(result_df)

Paso 5: Aplicar el control de acceso

Controlar quiénes pueden leer y escribir archivos en los volúmenes. Para más información sobre cómo administrar privilegios en el catálogo de Unity, consulte Administrar privilegios en el catálogo de Unity.

Paso 5.1: Conceder acceso

SQL

-- Replace <user-or-group-name> with your workspace group or user name

-- Grant read access
GRANT READ VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

-- Grant read and write access
GRANT READ VOLUME, WRITE VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

-- Grant all privileges
GRANT ALL PRIVILEGES ON VOLUME unstructured_data_lab.raw.files_volume
TO `<user-or-group-name>`;

Pitón

# Replace <user-or-group-name> with your workspace group or user name
spark.sql("""
    GRANT READ VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

spark.sql("""
    GRANT READ VOLUME, WRITE VOLUME ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

spark.sql("""
    GRANT ALL PRIVILEGES ON VOLUME unstructured_data_lab.raw.files_volume
    TO `<user-or-group-name>`
""")

Explorador de catálogos

1. Vaya a la pestaña Permisos de la página del volumen.
2. Haga clic en Conceder.
3. Escriba la dirección de correo electrónico de un usuario o el nombre de un grupo.
4. Seleccione los permisos que quiere conceder.
5. Haga clic en Confirmar.

Paso 5.2: Ver los privilegios actuales

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Pitón

display(spark.sql("SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume"))

Explorador de catálogos

La pestaña Permisos de la página de volúmenes muestra qué usuarios y grupos tienen acceso al volumen.

Paso 6: Configuración de la ingesta incremental

Utilice el cargador automático para procesar automáticamente los archivos nuevos a medida que llegan a su volumen. Este patrón es útil para flujos de trabajo continuos de ingesta de datos. Para obtener más patrones de ingesta, consulte Patrones comunes de carga de datos.

Paso 6.1: Creación de una tabla de streaming

SQL

CREATE OR REFRESH STREAMING TABLE document_ingestion
SCHEDULE EVERY 1 HOUR
AS SELECT
  path,
  modificationTime,
  length,
  content,
  _metadata,
  current_timestamp() AS ingestion_time
FROM STREAM(read_files(
  '/Volumes/unstructured_data_lab/raw/files_volume/incoming/',
  format => 'binaryFile'
));

Pitón

from pyspark.sql.functions import current_timestamp, col

dbutils.fs.mkdirs("/Volumes/unstructured_data_lab/raw/files_volume/incoming/")

df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("/Volumes/unstructured_data_lab/raw/files_volume/incoming/")

df_enriched = df \
    .withColumn("ingestion_time", current_timestamp()) \
    .withColumn("source_file", col("_metadata.file_path"))

query = df_enriched.writeStream \
    .option("checkpointLocation",
            "/Volumes/unstructured_data_lab/raw/files_volume/_checkpoints/docs") \
    .trigger(availableNow=True) \
    .toTable("document_ingestion")

query.awaitTermination()

Paso 7: Compartir archivos con OpenSharing

Comparta volúmenes de forma segura con los usuarios de otras organizaciones que usan OpenSharing. Debe crear un destinatario antes de compartirlo. Un destinatario representa una organización externa o un usuario que puede acceder a los datos compartidos. Consulte Creación de destinatarios de datos para OpenSharing (Uso compartido de Databricks a Databricks) para la configuración del destinatario.

Paso 7.1: Creación y configuración de un recurso compartido

SQL

-- Create a share
CREATE SHARE IF NOT EXISTS unstructured_data_share
COMMENT 'Document files for partners';

-- Add the volume
ALTER SHARE unstructured_data_share
ADD VOLUME unstructured_data_lab.raw.files_volume;

-- Create a recipient
CREATE RECIPIENT IF NOT EXISTS <partner_org>
USING ID '<recipient-sharing-identifier>';

-- Grant access
GRANT SELECT ON SHARE unstructured_data_share
TO RECIPIENT <partner_org>;

Pitón

spark.sql("""
    CREATE SHARE IF NOT EXISTS unstructured_data_share
    COMMENT 'Document files for partners'
""")

spark.sql("""
    ALTER SHARE unstructured_data_share
    ADD VOLUME unstructured_data_lab.raw.files_volume
""")

spark.sql("""
    CREATE RECIPIENT IF NOT EXISTS <partner_org>
    USING ID '<recipient-sharing-identifier>'
""")

spark.sql("""
    GRANT SELECT ON SHARE unstructured_data_share
    TO RECIPIENT <partner_org>
""")

Paso 7.2: Acceso a datos compartidos (como destinatario)

SQL

-- View available shares
SHOW SHARES IN PROVIDER <provider_name>;

-- Create a catalog from the share
CREATE CATALOG IF NOT EXISTS shared_documents
FROM SHARE <provider_name>.unstructured_data_share;

-- Query shared files
SELECT * EXCEPT (content), _metadata
FROM read_files(
  '/Volumes/shared_documents/raw/files_volume/',
  format => 'binaryFile'
)
LIMIT 10;

Pitón

spark.sql("SHOW SHARES IN PROVIDER <provider_name>").show()

spark.sql("""
    CREATE CATALOG IF NOT EXISTS shared_documents
    FROM SHARE <provider_name>.unstructured_data_share
""")

df = spark.read.format("binaryFile") \
    .load("/Volumes/shared_documents/raw/files_volume/")

df.select("path", "modificationTime", "length").show(10)

Paso 8: Limpieza de archivos

Quite los archivos cuando ya no sean necesarios.

Pitón

# Delete a single file
dbutils.fs.rm("/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf")

# Delete a directory recursively
dbutils.fs.rm("/Volumes/unstructured_data_lab/raw/files_volume/sample_files/", recurse=True)

Interfaz de línea de comandos (CLI)

# Delete a single file
databricks fs rm dbfs:/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf

# Delete a directory recursively
databricks fs rm -r dbfs:/Volumes/unstructured_data_lab/raw/files_volume/sample_files/

import os
os.remove("/Volumes/unstructured_data_lab/raw/files_volume/covid.pdf")

import shutil
shutil.rmtree("/Volumes/unstructured_data_lab/raw/files_volume/sample_files/")

Recursos adicionales

Continuar aprendiendo sobre volúmenes

• ¿Qué son los volúmenes del catálogo de Unity?
• Trabajar con archivos en volúmenes de Unity Catalog
• Creación y administración de volúmenes de Catálogo de Unity

Exploración de las funcionalidades relacionadas

• Búsqueda de contenido para volúmenes del Catálogo de Unity
• Enriquecimiento de datos mediante AI Functions
• Patrones comunes de carga de datos
• Uso compartido de datos y recursos de inteligencia artificial de forma segura

Referencias de función SQL

• read_files función con valores de tabla
• ai_parse_document Función