Trabalhar com dados não estruturados em volumes

Esta página mostra como armazenar, consultar e processar arquivos de dados não estruturados usando volumes do Catálogo do Unity. Você aprenderá a carregar arquivos, consultar metadados, processar arquivos com funções de IA, aplicar controle de acesso e compartilhar volumes com outras organizações. Sempre que possível, as instruções para trabalhar neste tutorial usando a interface do usuário do Catalog Explorer foram incluídas. Se nenhuma opção do Gerenciador de Catálogos for mostrada, use os comandos Python ou SQL fornecidos.

Para obter uma visão geral completa dos recursos de volume e casos de uso, consulte O que são volumes do Catálogo do Unity?.

Requirements

• Um workspace do Azure Databricks com o Unity Catalog habilitado.
• CREATE CATALOG privilégio no metastore. Confira Criar catálogos. Se você não puder criar um catálogo, peça acesso ao administrador ou use um catálogo existente em que você tenha o CREATE SCHEMA privilégio.
• Databricks Runtime 14.3 LTS e versões posteriores.
• Para funções de IA: um espaço de trabalho em uma região com suporte.
• Para OpenSharing: privilégios CREATE SHARE e CREATE RECIPIENT no metastore. Consulte os dados do Share e os ativos de IA com segurança.

Etapa 1: Criar um volume

Crie um catálogo, um esquema e um volume para armazenar seus arquivos. Para obter instruções detalhadas de gerenciamento de volume, consulte Criar e gerenciar volumes do Catálogo do Unity.

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

Gerenciador de Catálogos

1. Clique no Catálogo na barra lateral.
2. Clique em Criar>um catálogo.
3. Insira unstructured_data_lab como o nome do catálogo.
4. Clique em Criar.
5. Clique em Exibir catálogo.

Na página do catálogo:

1. Clique em Criar esquema.
2. Insira bruto como Nome do esquema.
3. Clique em Criar.

Etapa 1.2: Criar um volume gerenciado

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

Gerenciador de Catálogos

Na página do esquema:

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

Etapa 2: Carregar arquivos

Carregue arquivos para o seu volume. Para obter exemplos abrangentes de gerenciamento de arquivos, consulte Trabalhar com arquivos em volumes do Catálogo do Unity.

Etapa 2.1: Carregar arquivos

Você pode usar exemplos de databricks-datasets para este tutorial ou carregar seus próprios arquivos usando a interface de usuário do Catalog Explorer.

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

Gerenciador de Catálogos

O código Python na guia Python carrega dois arquivos (um JPG e um PDF) e um diretório que inclui .txt e .csv arquivos. Para carregar arquivos usando o Gerenciador de Catálogos:

1. Na página de volume, clique em Carregar para este volume.
2. Na caixa de diálogo Carregar arquivos, em Arquivos, clique em procurar ou arraste e solte arquivos na área para arrastar e soltar.
3. Em Volume de destino, verifique se o volume que você criou na etapa anterior está selecionado.

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

Gerenciador de Catálogos

Quando os arquivos são carregados, eles aparecem na página de volume. Clique em um nome de arquivo para ver uma visualização ou clique em um diretório para exibir arquivos individuais.

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

Etapa 3: Consultar metadados de arquivo

Consulte as informações do arquivo para entender o conteúdo do seu volume. Para obter mais padrões de consulta, consulte Listar e consultar arquivos em volumes com SQL.

Etapa 3.1: Mostrar metadados de arquivo

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)

Gerenciador de Catálogos

A página de volume no Gerenciador de Catálogos mostra o Nome de cada arquivo (incluindo a extensão), o tamanho e a data da última modificação .

Etapa 4: Consultar e processar arquivos

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

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

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

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

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

Etapa 4.3: Filtrar e analisar por nome de arquivo

Este exemplo filtra arquivos de imagem com a subcadeia de caracteres "rose" em seu nome de arquivo.

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)

Etapa 4.4: Unir arquivos com tabelas estruturadas

Este exemplo utiliza números de linha para associar arquivos a viagens de táxi, para fins de demonstração. Em produção, realize junções com base 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)

Etapa 5: Aplicar controle de acesso

Controlar quem pode ler e gravar arquivos em seus volumes. Para saber mais sobre como gerenciar privilégios no Catálogo do Unity, consulte Gerenciar privilégios no Catálogo do Unity.

Etapa 5.1: Conceder acesso

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

Gerenciador de Catálogos

1. Vá para a guia Permissões na página de volume.
2. Selecione Conceder.
3. Insira o endereço de email para um usuário ou o nome de um grupo.
4. Selecione as permissões a serem concedidas.
5. Clique em Confirmar.

Etapa 5.2: Exibir 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"))

Gerenciador de Catálogos

A guia Permissões na página de volume mostra quais usuários e grupos têm acesso ao volume.

Etapa 6: Configurar a ingestão incremental

Use o Carregador Automático para processar automaticamente novos arquivos conforme eles chegam ao volume. Esse padrão é útil para fluxos de trabalho de ingestão contínua de dados. Para obter mais padrões de ingestão, consulte padrões comuns de carregamento de dados.

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

Etapa 7: Compartilhar arquivos com o OpenSharing

Compartilhe volumes com segurança com usuários de outras organizações usando o OpenSharing. Você deve criar um destinatário antes de compartilhar. Um destinatário representa uma organização externa ou um usuário que pode acessar seus dados compartilhados. Consulte Criar destinatários de dados para o OpenSharing (compartilhamento de Databricks para Databricks) para configurar os destinatários.

Etapa 7.1: Criar e configurar um compartilhamento

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

Etapa 7.2: Acessar dados compartilhados (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)

Etapa 8: Limpar arquivos

Remova os arquivos quando eles não forem mais 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/

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 aprendendo sobre volumes

• O que são os volumes do Catálogo do Unity?
• Trabalhar com arquivos em volumes do Catálogo do Unity
• Criar e gerenciar volumes do Catálogo do Unity

Explorar funcionalidades relacionadas

• Busca de conteúdo por volumes do Catálogo Unity
• Enriquecer dados usando o AI Functions
• Padrões comuns de carregamento de dados
• Compartilhar dados e ativos de IA com segurança

Referências de função SQL

• read_files Função com valor de tabela
• ai_parse_document função