Trabalhar com dados não estruturados em volumes

Esta página mostra-lhe como armazenar, consultar e processar ficheiros de dados não estruturados usando volumes do Unity Catalog. Vai aprender a carregar ficheiros, consultar metadados, processar ficheiros com funções de IA, aplicar controlo de acesso e partilhar volumes com outras organizações. Sempre que possível, foram incluídas instruções para realizar este tutorial usando a interface do Explorador de Catálogos. Se não aparecer a opção do Explorador de Catálogos , use os comandos Python ou SQL fornecidos.

Para uma visão geral completa das capacidades e casos de uso dos volumes, consulte O que são volumes do Unity Catalog?.

Observação

Este tutorial utiliza funções de IA para processar ficheiros por caminho. Disponível em Beta, o FILE tipo permite armazenar referências de ficheiros e metadados como valores de coluna numa tabela. Ver tipo de ficheiro e dados não estruturados.

Requerimentos

  • Um espaço de trabalho Azure Databricks com o Unity Catalog ativado.
  • CREATE CATALOG Privilégio em Metastore. Ver Criar catálogos. Se não conseguires criar um catálogo, pede acesso ao teu administrador ou usa um catálogo existente onde tens esse CREATE SCHEMA privilégio.
  • Databricks Runtime 14.3 LTS e superiores.
  • Para funções de IA: Um espaço de trabalho numa região suportada.
  • Para OpenSharing: CREATE SHARE e CREATE RECIPIENT privilégios na metastore. Veja Partilhar dados e ativos de IA de forma segura.

Passo 1: Criar um volume

Cria um catálogo, esquema e volume para armazenar os teus ficheiros. Para instruções detalhadas de gestão de volumes, consulte Criar e gerir volumes do Catálogo Unity.

Passo 1.1: Criar um catálogo e um 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;

Python

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. Clique no ícone Dados.Catálogo na barra lateral.
  2. Clique em Criar>um catálogo.
  3. Introduza unstructured_data_lab como nome do Catálogo.
  4. Clique em Criar.
  5. Clique em Ver catálogo.

Na página do catálogo:

  1. Clique em Criar esquema.
  2. Introduza raw como nome do Esquema.
  3. Clique em Criar.

Passo 1.2: Criar um volume gerido

SQL

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

Python

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

Explorador de Catálogos

Na página do esquema:

  1. Clique em Criar>Volume.
  2. Insira files_volume como nome do Volume.
  3. Verifique se Managed volume está selecionado.
  4. Clique em Criar.

Passo 2: Carregar ficheiros

Carrega ficheiros para o teu volume. Para exemplos abrangentes de gestão de ficheiros, consulte Trabalhar com ficheiros em volumes do Unity Catalog.

Passo 2.1: Carregar ficheiros

Podes usar exemplos deste databricks-datasets tutorial, ou carregar os teus próprios ficheiros usando a interface do Explorador de Catálogos.

Observação

Podes usar os comandos Python para copiar ficheiros databricks-datasets para o teu volume mesmo que não estejas familiarizado com Python. Consulte Gerenciar cadernos Databricks para instruções sobre como executar comandos em cadernos.

Python

# 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

O código Python no separador Python carrega dois ficheiros (um JPG e um PDF) e um diretório que inclui ficheiros .txt e .csv. Para carregar ficheiros usando o Explorador de Catálogo:

  1. Na página do volume, clique em Carregar para este volume.
  2. No diálogo Carregar ficheiros, em Ficheiros, clique em navegar ou arrastar e largar ficheiros para a área de largada.
  3. Em Volume de Destino, verifique se o volume que criou na etapa anterior está selecionado.

Passo 2.2: Verificar o upload

SQL

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

Python

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

Quando os ficheiros são carregados, aparecem na página do volume. Clique num nome de ficheiro para ver uma pré-visualização, ou clique num diretório para ver ficheiros individuais.

Alternativa: Usar o comando mágico %fs

Use o %fs comando mágico:

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

Passo 3: Consultar metadados do ficheiro

Consulte a informação do ficheiro para compreender o conteúdo do volume. Para mais padrões de consulta, veja Listar e consultar ficheiros em volumes com SQL.

Passo 3.1: Mostrar metadados do ficheiro

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'
);

Python

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

A página de volume no Explorador de Catálogo mostra o Nome de cada ficheiro (incluindo a extensão), o Tamanho e a data da última modificação .

Passo 4: Consultar e processar ficheiros

Use as funções de IA do Azure Databricks para extrair conteúdo de documentos e analisar imagens. Para uma visão geral completa das capacidades das funções de IA, consulte Enriquecer dados usando Funções de IA.

Observação

As funções de IA requerem um espaço de trabalho numa região suportada. Veja Enriquecer dados usando Funções de IA.

Se não tiveres acesso a funções de IA, usa bibliotecas Python padrão em vez disso. Expanda as secções Alternativas abaixo para exemplos.

Passo 4.1: Analisar 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'
);

Python

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)
Alternativa: Analisar PDFs sem funções de IA

Se as funções de IA não estiverem disponíveis na sua região, use bibliotecas em Python:

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

Passo 4.2: Analisar imagens

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;

Python

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)
Alternativa: extrair metadados de imagem sem funções de IA

Para extrair metadados de imagem sem funções de IA:

%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.*"))

Passo 4.3: Filtrar e analisar por nome de ficheiro

Este exemplo filtra ficheiros de imagem com a substring "rose" no nome do ficheiro.

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%';

Python

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)

Passo 4.4: Juntar ficheiros com tabelas estruturadas

Este exemplo utiliza números de linha para associar ficheiros a viagens de táxi com o objetivo de demonstração. Na produção, participe em chaves de negócio 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;

Python

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)

Passo 5: Aplicar controlo de acesso

Controla quem pode ler e escrever ficheiros nos teus volumes. Para saber mais sobre a gestão de privilégios no Catálogo Unity, consulte Gerir privilégios no Catálogo Unity.

Passo 5.1: Acesso a subsídios

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>`;

Python

# 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. Vai ao separador Permissões na página do volume.
  2. Clique em Conceder.
  3. Insira o endereço de e-mail de um usuário ou o nome de um grupo.
  4. Selecione as permissões a serem concedidas.
  5. Clique em Confirmar.

Passo 5.2: Ver privilégios atuais

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Python

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

Explorador de Catálogos

O separador Permissões na página do volume mostra quais os utilizadores e grupos que têm acesso ao volume.

Passo 6: Configurar a ingestão incremental

Use o Auto Loader para processar automaticamente novos ficheiros à medida que chegam ao seu volume. Este padrão é útil para fluxos de trabalho contínuos de ingestão de dados. Para mais padrões de ingestão, veja Padrões comuns de carregamento de dados.

Passo 6.1: Criar uma tabela 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'
));

Python

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

Passo 7: Partilhar ficheiros com o OpenSharing

Partilhe volumes de forma segura com utilizadores de outras organizações que utilizam o OpenSharing. Deve criar um destinatário antes de partilhar. Um destinatário representa uma organização ou utilizador externo que pode aceder aos seus dados partilhados. Consulte Criar destinatários de dados para OpenSharing (partilha entre Databricks) para configurar o destinatário.

Passo 7.1: Criar e configurar uma partilha

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>;

Python

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

Passo 7.2: Aceder a dados partilhados (como destinatário)

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;

Python

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)

Passo 8: Limpar ficheiros

Remova ficheiros quando já não forem necessários.

Python

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

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/
Alternativa: Usar Python padrão 
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 adicionais

Continuar a aprender sobre volumes

Referências de funções SQL

  • Last updated on