Arbeiten mit unstrukturierten Daten in Volumes

Auf dieser Seite erfahren Sie, wie Sie unstrukturierte Datendateien mithilfe von Unity-Katalogvolumes speichern, abfragen und verarbeiten. Sie erfahren, wie Sie Dateien, Abfragemetadaten, Prozessdateien mit KI-Funktionen hochladen, Zugriffssteuerung anwenden und Volumes für andere Organisationen freigeben. Wenn möglich, wurden Anweisungen zum Arbeiten in diesem Lernprogramm mit der Benutzeroberfläche des Katalog-Explorers enthalten. Wenn keine Katalog-Explorer-Option angezeigt wird, verwenden Sie die bereitgestellten Python- oder SQL-Befehle.

Eine vollständige Übersicht über Volumenfunktionen und Anwendungsfälle finden Sie unter "Was sind Unity-Katalogvolumes?".

Anforderungen

• Ein Azure Databricks-Arbeitsbereich mit aktiviertem Unity-Katalog.
• CREATE CATALOG-Privileg für den Metastore. Siehe Erstellen von Katalogen. Wenn Sie keinen Katalog erstellen können, bitten Sie Ihren Administrator um Zugriff, oder verwenden Sie einen vorhandenen Katalog, in dem Sie über die CREATE SCHEMA Berechtigung verfügen.
• Databricks Runtime 14.3 LTS und höher.
• Für KI-Funktionen: Ein Arbeitsbereich in einer unterstützten Region.
• Für OpenSharing: CREATE SHARE und CREATE RECIPIENT Berechtigungen im Metastore. Weitere Informationen hierzu finden Sie unter Sicheres Freigeben von Daten- und KI-Ressourcen.

Schritt 1: Erstellen eines Volumes

Erstellen Sie einen Katalog, ein Schema und ein Volume, um Ihre Dateien zu speichern. Ausführliche Anweisungen zur Volumenverwaltung finden Sie unter Erstellen und Verwalten von Unity-Katalogvolumes.

Schritt 1.1: Erstellen eines Katalogs und Schemas

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

Katalog-Explorer

1. Klicken Sie auf Katalog in der Randleiste.
2. Klicken Sie auf ">".
3. Geben Sie unstructured_data_lab als Katalognamen ein.
4. Klicken Sie auf "Erstellen".
5. Klicken Sie auf "Katalog anzeigen".

Auf der Katalogseite:

1. Klicken Sie auf "Schema erstellen".
2. Geben Sie raw als Schema Name ein.
3. Klicken Sie auf "Erstellen".

Schritt 1.2: Erstellen eines verwalteten Volumes

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

Katalog-Explorer

Auf der Schemaseite:

1. Klicken Sie auf "Erstellen>Volume".
2. Geben Sie files_volume als Volumename ein.
3. Stellen Sie sicher, dass Verwaltetes Volume ausgewählt ist.
4. Klicken Sie auf "Erstellen".

Schritt 2: Hochladen von Dateien

Laden Sie Dateien auf Ihr Volume hoch. Umfassende Dateiverwaltungsbeispiele finden Sie unter "Arbeiten mit Dateien in Unity-Katalogvolumes".

Schritt 2.1: Hochladen von Dateien

Sie können Beispiele von databricks-datasets für dieses Lernprogramm verwenden oder Ihre eigenen Dateien über die Benutzeroberfläche des Katalog-Explorers hochladen.

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

Katalog-Explorer

Der Python-Code auf der Registerkarte Python lädt zwei Dateien (eine JPG und eine PDF) sowie ein Verzeichnis hoch, das .txt- und .csv-Dateien enthält. So laden Sie Dateien mit dem Katalog-Explorer hoch:

1. Klicken Sie auf der Volumeseite auf "Auf dieses Volume hochladen".
2. Klicken Sie im Dialogfeld "Dateien hochladen " unter "Dateien" auf "Durchsuchen ", oder ziehen Sie Dateien in die Dropdownzone.
3. Überprüfen Sie unter Zielvolume, ob das volume, das Sie im vorherigen Schritt erstellt haben, ausgewählt ist.

Schritt 2.2: Überprüfen des Uploads

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

Katalog-Explorer

Wenn Dateien hochgeladen werden, werden sie auf der Volumeseite angezeigt. Klicken Sie auf einen Dateinamen, um eine Vorschau anzuzeigen, oder klicken Sie auf ein Verzeichnis, um einzelne Dateien anzuzeigen.

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

Schritt 3: Abfragedateimetadaten

Dateiinformationen abfragen, um zu verstehen, was sich in Ihrem Volume befindet. Weitere Abfragemuster finden Sie unter Listen- und Abfragedateien in Volumes mit SQL.

Schritt 3.1: Anzeigen von Dateimetadaten

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)

Katalog-Explorer

Auf der Volumeseite im Katalog-Explorer werden der Name der einzelnen Dateien (einschließlich Erweiterung), Größe und Datum der letzten Änderung angezeigt .

Schritt 4: Abfragen und Verarbeiten von Dateien

Verwenden Sie Azure Databricks AI-Funktionen, um Inhalte aus Dokumenten zu extrahieren und Bilder zu analysieren. Eine vollständige Übersicht über KI-Funktionen finden Sie unter Anreichern von Daten mithilfe von KI-Funktionen.

Schritt 4.1: Analysieren von Dokumenten

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

Schritt 4.2: Analysieren von Bildern

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

Schritt 4.3: Filtern und Analysieren nach Dateinamen

In diesem Beispiel wird nach Bilddateien mit der Teilzeichenfolge "rose" im Dateinamen gefiltert.

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)

Schritt 4.4: Verknüpfen von Dateien mit strukturierten Tabellen

In diesem Beispiel werden Zeilennummern verwendet, um Dateien mit Taxifahrten zu Demonstrationszwecken zu koppeln. Stellen Sie in der Produktion eine Verbindung zu aussagekräftigen Geschäftsschlüsseln her.

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)

Schritt 5: Anwenden der Zugriffssteuerung

Steuern Sie, wer Dateien in Ihren Volumes lesen und schreiben kann. Weitere Informationen zum Verwalten von Berechtigungen im Unity-Katalog finden Sie unter Verwalten von Berechtigungen im Unity-Katalog.

Schritt 5.1: Gewähren des Zugriffs

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

Katalog-Explorer

1. Wechseln Sie auf der Volumeseite zur Registerkarte Berechtigungen.
2. Klicken Sie auf Gewähren.
3. Geben Sie die E-Mail-Adresse für einen Benutzer oder den Namen einer Gruppe ein.
4. Wählen Sie die zu erteilenden Berechtigungen aus.
5. Klicken Sie auf Bestätigen.

Schritt 5.2: Anzeigen aktueller Berechtigungen

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

Katalog-Explorer

Auf der Registerkarte "Berechtigungen " auf der Volumeseite wird angezeigt, welche Benutzer und Gruppen Zugriff auf das Volume haben.

Schritt 6: Einrichten der inkrementellen Erfassung

Verwenden Sie das automatische Laden, um neue Dateien automatisch zu verarbeiten, sobald sie in Ihrem Volume ankommen. Dieses Muster eignet sich für fortlaufende Datenaufnahmeworkflows. Weitere Einspeisemuster finden Sie unter "Allgemeine Datenlademuster".

Schritt 6.1: Erstellen einer Streamingtabelle

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

Schritt 7: Freigeben von Dateien mit OpenSharing

Teilen Sie Volumes sicher mit Benutzern in anderen Organisationen, die OpenSharing verwenden. Sie müssen einen Empfänger erstellen, bevor Sie freigeben. Ein Empfänger stellt eine externe Organisation oder einen Benutzer dar, der auf Ihre freigegebenen Daten zugreifen kann. Informationen zum Einrichten von Empfängern finden Sie unter Erstellen von Datenempfängern für OpenSharing (Databricks-zu-Databricks-Freigabe).

Schritt 7.1: Erstellen und Konfigurieren einer Freigabe

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

Schritt 7.2: Zugreifen auf freigegebene Daten (als Empfänger)

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)

Schritt 8: Bereinigen von Dateien

Entfernen Sie Dateien, wenn sie nicht mehr benötigt werden.

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

Weitere Ressourcen

Weitere Informationen zu Volumes

• Was sind Unity Catalog Volumes?
• Arbeiten mit Dateien in Unity-Katalogvolumes
• Erstellen und Verwalten von Unity-Katalogvolumes

Erkunden verwandter Funktionen

• Inhaltssuche für Unity-Katalog-Bände
• Anreichern von Daten mithilfe von KI-Funktionen
• Allgemeine Datenlademuster
• Sicheres Freigeben von Daten und KI-Ressourcen

SQL-Funktionsverweise

• read_filesTabellenwertfunktion
• ai_parse_document Funktion