Werken met ongestructureerde gegevens in volumes

Op deze pagina ziet u hoe u ongestructureerde gegevensbestanden opslaat, opvraagt en verwerkt met behulp van Unity Catalog-volumes. U leert hoe u bestanden uploadt, metagegevens opvraagt, bestanden verwerkt met AI-functies, toegangsbeheer toepast en volumes deelt met andere organisaties. Waar mogelijk zijn de instructies voor het uitvoeren van deze zelfstudie met behulp van de gebruikersinterface van Catalog Explorer opgenomen. Als er geen Catalogusverkenner-optie wordt weergegeven, gebruikt u de opgegeven Python- of SQL-opdrachten.

Zie Wat zijn Unity Catalog-volumes? voor een volledig overzicht van volumemogelijkheden en use cases.

Requirements

• Een Azure Databricks-werkruimte waarvoor Unity Catalog is ingeschakeld.
• CREATE CATALOG bevoegdheid voor de metastore. Zie Catalogi maken. Als u geen catalogus kunt maken, vraagt u uw beheerder om toegang of gebruikt u een bestaande catalogus met de CREATE SCHEMA bevoegdheid.
• Databricks Runtime 14.3 LTS en hoger.
• Voor AI-functies: een werkruimte in een ondersteunde regio.
• Voor OpenSharing: CREATE SHARE en CREATE RECIPIENT bevoegdheden voor de metastore. Zie Gegevens en AI-assets veilig delen.

Stap 1: Een volume maken

Maak een catalogus, schema en volume om uw bestanden op te slaan. Zie Unity Catalog-volumes maken en beheren voor gedetailleerde instructies voor volumebeheer.

Stap 1.1: Een catalogus en schema maken

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

Catalogusverkenner

1. Klik op Catalogus in de zijbalk.
2. Klik opEen catalogus maken>.
3. Voer unstructured_data_lab in als catalogusnaam.
4. Klik op Create.
5. Klik op Catalogus weergeven.

Op de cataloguspagina:

1. Klik op Schema maken.
2. Voer onbewerkt in als Schema-naam.
3. Klik op Create.

Stap 1.2: Een beheerd volume maken

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

Catalogusverkenner

Op de schemapagina:

1. Klik op Maken>Volume.
2. Voer files_volume in als volumenaam.
3. Controleer of het beheerde volume is geselecteerd.
4. Klik op Create.

Stap 2: Bestanden uploaden

Bestanden uploaden naar uw volume. Zie Werken met bestanden in Unity Catalog-volumes voor uitgebreide voorbeelden van bestandsbeheer.

Stap 2.1: Bestanden uploaden

U kunt voorbeelden uit databricks-datasets deze zelfstudie gebruiken of uw eigen bestanden uploaden met behulp van de gebruikersinterface van 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}")

Catalogusverkenner

De Python-code op het Python-tabblad uploadt twee bestanden (jpg en pdf) en een map met .txt bestanden en .csv bestanden. Bestanden uploaden met Catalog Explorer:

1. Klik op de volumepagina op Uploaden naar dit volume.
2. Klik in het dialoogvenster Bestanden uploaden onder Bestanden op bladeren of sleep bestanden naar de dropzone.
3. Controleer onder Doelvolume of het volume dat u in de vorige stap hebt gemaakt, is geselecteerd.

Stap 2.2: De upload controleren

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

Catalogusverkenner

Wanneer bestanden worden geüpload, worden ze weergegeven op de volumepagina. Klik op een bestandsnaam om een voorbeeld te bekijken of klik op een map om afzonderlijke bestanden weer te geven.

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

Stap 3: Metagegevens van bestand opvragen

Voer een query uit op bestandsgegevens om te begrijpen wat er in uw volume staat. Zie Lijst en query's uitvoeren op bestanden in volumes met SQL voor meer querypatronen.

Stap 3.1: Bestandsmetagegevens weergeven

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)

Catalogusverkenner

Op de volumepagina in Catalog Explorer wordt de naam van elk bestand (inclusief de extensie), de grootte en de datum van laatste wijziging weergegeven.

Stap 4: Query' en procesbestanden uitvoeren

Azure Databricks AI-functies gebruiken om inhoud uit documenten te extraheren en afbeeldingen te analyseren. Voor een volledig overzicht van de mogelijkheden van AI-functies, zie Transform unstructured data with AI Functions.

Stap 4.1: Documenten parseren

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

Stap 4.2: Afbeeldingen analyseren

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

Stap 4.3: Filteren en analyseren op bestandsnaam

In dit voorbeeld worden afbeeldingsbestanden gefilterd met de subtekenreeks 'rose' in hun bestandsnaam.

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)

Stap 4.4: Bestanden samenvoegen met gestructureerde tabellen

In dit voorbeeld worden rijnummers gebruikt om bestanden te koppelen aan taxiritten voor demonstratiedoeleinden. In productie, combineer gegevens met zinvolle bedrijfssleutels.

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)

Stap 5: Toegangsbeheer toepassen

Bepalen wie bestanden in uw volumes kan lezen en schrijven. Zie Bevoegdheden beheren in Unity Catalog voor meer informatie over het beheren van bevoegdheden in Unity Catalog.

Stap 5.1: Toegang verlenen

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

Catalogusverkenner

1. Ga naar het tabblad Machtigingen op de volumepagina.
2. Klik op Toestaan.
3. Voer het e-mailadres in voor een gebruiker of de naam van een groep.
4. Selecteer de machtigingen die u wilt verlenen.
5. Klik op Bevestigen.

Stap 5.2: Huidige bevoegdheden weergeven

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

Catalogusverkenner

Op het tabblad Machtigingen op de volumepagina ziet u welke gebruikers en groepen toegang hebben tot het volume.

Stap 6: Incrementele gegevensverwerking instellen

Gebruik Automatisch laden om automatisch nieuwe bestanden te verwerken wanneer ze binnenkomen in uw volume. Dit patroon is handig voor werkstromen voor continue gegevensopname. Zie Algemene patronen voor het laden van gegevens voor meer opnamepatronen.

Stap 6.1: Een streamingtabel maken

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

Stap 7: Bestanden delen met OpenSharing

Deel volumes veilig met gebruikers in andere organisaties met behulp van OpenSharing. U moet een geadresseerde maken voordat u het deelt. Een ontvanger vertegenwoordigt een externe organisatie of gebruiker die toegang heeft tot uw gedeelde gegevens. Zie Gegevensontvangers maken voor OpenSharing (Databricks-naar-Databricks-deling) voor de configuratie van ontvangers.

Stap 7.1: Een share maken en configureren

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

Stap 7.2: Toegang krijgen tot gedeelde gegevens (als ontvanger)

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)

Stap 8: Bestanden opschonen

Verwijder bestanden wanneer ze niet meer nodig zijn.

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 (Command Line Interface)

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

Aanvullende informatiebronnen

Doorgaan met leren over volumes

• Wat zijn Unity Catalog-volumes?
• Werken met bestanden in Unity Catalog-volumes
• Unity Catalog-volumes maken en beheren

Verwante mogelijkheden verkennen

• Inhoudszoektocht naar Unity Catalog-volumes
• Transformeer ongestructureerde data met behulp van AI-functies
• Algemene patronen voor het laden van gegevens
• Gegevens en AI-assets veilig delen

Sql-functieverwijzingen

• read_files tabelgewaardeerde functie
• ai_parse_document Functie