Práce s nestrukturovanými daty ve svazcích

Na této stránce se dozvíte, jak ukládat, dotazovat a zpracovávat nestrukturované datové soubory pomocí svazků katalogu Unity. Naučíte se nahrávat soubory, dotazovat metadata, zpracovávat soubory pomocí funkcí AI, používat řízení přístupu a sdílet svazky s jinými organizacemi. Pokud je to možné, byly zahrnuty pokyny pro práci v tomto kurzu pomocí uživatelského rozhraní Průzkumníka katalogů. Pokud se nezobrazí žádná možnost Průzkumníka katalogu , použijte zadané příkazy Pythonu nebo SQL.

Úplný přehled možností svazků a případů použití najdete v tématu Co jsou svazky katalogu Unity?.

Poznámka:

Tento tutoriál využívá AI funkce ke zpracování souborů podle cesty. Dostupný v beta verzi, tento FILE typ umožňuje ukládat odkazy na soubory a metadata jako sloupce v tabulce. Viz typ souboru a nestrukturovaná data.

Požadavky

• Pracovní prostor Azure Databricks s povoleným Unity Catalog.
• CREATE CATALOG oprávnění pro metastore. Viz Vytváření katalogů. Pokud nemůžete vytvořit katalog, požádejte správce o přístup nebo použijte existující katalog, ve kterém máte CREATE SCHEMA oprávnění.
• Databricks Runtime 14.3 LTS a novější.
• Pro funkce AI: Pracovní prostor v podporované oblasti.
• Pro OpenSharing: oprávnění CREATE SHARE a CREATE RECIPIENT pro metastore. Informace o bezpečném sdílení dat a prostředků AI

Krok 1: Vytvoření svazku

Vytvořte katalog, schéma a svazek pro ukládání souborů. Podrobné pokyny ke správě svazků najdete v tématu Vytvoření a správa svazků katalogu Unity.

Krok 1.1: Vytvoření katalogu a schématu

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

Průzkumník katalogu

1. Klikněte na Katalog na bočním panelu
2. Klikněte na Vytvořit>katalog.
3. Jako název katalogu zadejte unstructured_data_lab.
4. Klikněte na Vytvořit.
5. Klikněte na Zobrazit katalog.

Na stránce katalogu:

1. Klikněte na Vytvořit schéma.
2. Jako název schématu zadejte nezpracovaný název.
3. Klikněte na Vytvořit.

Krok 1.2: Vytvoření spravovaného svazku

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

Průzkumník katalogu

Na stránce schématu:

1. Klikněte na Vytvořit>Svazek.
2. Jako název svazku zadejte files_volume.
3. Ověřte, že je vybraný řízený svazek.
4. Klikněte na Vytvořit.

Krok 2: Nahrání souborů

Nahrajte soubory na svazek. Komplexní příklady správy souborů najdete v tématu Práce se soubory ve svazcích katalogu Unity.

Krok 2.1: Nahrání souborů

Příklady z databricks-datasets tohoto kurzu můžete použít nebo nahrát vlastní soubory pomocí uživatelského rozhraní Průzkumníka katalogů.

Poznámka:

Příkazy Pythonu můžete použít ke kopírování souborů na váš svazek ze databricks-datasets, i když Python neznáte. Pokyny ke spouštění příkazů v poznámkových blocích najdete v tématu Správa poznámkových bloků Databricks .

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

Průzkumník katalogu

Kód Pythonu na kartě Python nahraje dva soubory (JPG a PDF) a adresář, který obsahuje .txt a .csv soubory. Nahrání souborů pomocí Průzkumníka katalogu:

1. Na stránce svazku klikněte na Nahrát na tento svazek.
2. V dialogovém okně Nahrát soubory klikněte v části Soubory na procházet nebo přetáhněte soubory do zóny přetažení.
3. V části Cílový svazek ověřte, že je vybraný svazek, který jste vytvořili v předchozím kroku.

Krok 2.2: Ověření nahrávání

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

Průzkumník katalogu

Když se soubory nahrají, zobrazí se na stránce svazku. Kliknutím na název souboru zobrazíte náhled nebo kliknutím na adresář zobrazíte jednotlivé soubory.

Alternativní: Použití příkazu magic %fs

Použijte příkaz %fs magický:

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

Krok 3: Dotazování metadat souboru

Zadejte dotaz na informace o souborech, abyste pochopili, co je ve vašem svazku. Další vzory dotazování najdete v tématu Výpis a dotazování souborů ve svazcích pomocí SQL.

Krok 3.1: Zobrazení metadat souborů

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)

Průzkumník katalogu

Stránka svazku v Průzkumníku katalogu zobrazuje název každého souboru (včetně přípony), velikost a datum poslední změny .

Krok 4: Dotazování a zpracování souborů

Pomocí funkcí AI Azure Databricks můžete extrahovat obsah z dokumentů a analyzovat obrázky. Úplný přehled funkcí umělé inteligence najdete v tématu Rozšiřování dat pomocí funkcí AI.

Poznámka:

Funkce AI vyžadují pracovní prostor v podporované oblasti. Viz Rozšíření dat pomocí funkcí AI.

Pokud nemáte přístup k funkcím AI, použijte místo toho standardní knihovny Pythonu. Rozbalte níže uvedené alternativní části pro příklady.

Krok 4.1: Analýza dokumentů

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)

Alternativní: Analýza souborů PDF bez funkcí AI

Pokud funkce AI nejsou ve vaší oblasti dostupné, použijte knihovny Pythonu:

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

Krok 4.2: Analýza obrázků

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: Extrahování metadat obrázků bez funkcí umělé inteligence

Extrahování metadat obrázků bez funkcí umělé inteligence:

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

Krok 4.3: Filtrování a analýza podle názvu souboru

Tento příklad filtruje soubory obrázků s podřetězcem "rose" v názvu souboru.

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)

Krok 4.4: Spojení souborů se strukturovanými tabulkami

Tento příklad používá čísla řádků ke spárování souborů s taxislužbami pro demonstrační účely. V produkčním prostředí se připojte ke smysluplným obchodním klíčům.

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)

Krok 5: Použití řízení přístupu

Určete, kdo může číst a zapisovat soubory ve vašich svazcích. Další informace o správě oprávnění v katalogu Unity najdete v tématu Správa oprávnění v katalogu Unity.

Krok 5.1: Udělení přístupu

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

Průzkumník katalogu

1. Přejděte na kartu Oprávnění na stránce svazku.
2. Klikněte na Povolit.
3. Zadejte e-mailovou adresu uživatele nebo jména skupiny.
4. Vyberte oprávnění, která chcete udělit.
5. Klikněte na tlačítko Potvrdit.

Krok 5.2: Zobrazení aktuálních oprávnění

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

Průzkumník katalogu

Karta Oprávnění na stránce svazku ukazuje, kteří uživatelé a skupiny mají k svazku přístup.

Krok 6: Nastavení přírůstkového příjmu dat

Pomocí nástroje Auto Loader můžete automaticky zpracovávat nové soubory po jejich doručení do svazku. Tento model je užitečný pro pracovní postupy průběžného příjmu dat. Další vzory příjmu dat najdete v tématu Běžné vzory načítání dat.

Krok 6.1: Vytvoření tabulky streamování

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

Krok 7: Sdílení souborů pomocí OpenSharingu

Sdílejte svazky bezpečně s uživateli v jiných organizacích pomocí OpenSharingu. Před sdílením musíte vytvořit příjemce. Příjemce představuje externí organizaci nebo uživatele, který má přístup k vašim sdíleným datům. Informace o nastavení příjemce najdete v tématu Vytváření příjemců dat pro OpenSharing (sdílení Databricks-to-Databricks).

Krok 7.1: Vytvoření a konfigurace sdílené složky

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

Krok 7.2: Přístup ke sdíleným datům (jako příjemce)

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)

Krok 8: Vyčištění souborů

Odeberte soubory, když už nejsou potřeba.

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/

Alternativní: Použití standardního Pythonu

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

Další zdroje informací

Další informace o svazcích

• Co jsou svazky katalogu Unity?
• Práce se soubory ve svazcích katalogu Unity
• Vytváření a správa svazků katalogu Unity

Prozkoumání souvisejících funkcí

• Vyhledávání obsahu svazků Unity Catalog
• Obohacení dat pomocí funkcí AI
• Běžné vzory načítání dat
• Bezpečné sdílení dat a prostředků AI

Odkazy na funkce SQL

• read_files tabulkově hodnotová funkce
• ai_parse_document funkce