Strukturálatlan adatok használata kötetekben

Ez a lap bemutatja, hogyan tárolhat, kérdezhet le és dolgozhat fel strukturálatlan adatfájlokat Unity Catalog-kötetek használatával. Megtudhatja, hogyan tölthet fel fájlokat, kérdezhet le metaadatokat, hogyan dolgozhat fel fájlokat AI-függvényekkel, hogyan alkalmazhatja a hozzáférés-vezérlést, és hogyan oszthat meg köteteket más szervezetekkel. Ahol lehetséges, az oktatóanyag a Katalóguskezelő felhasználói felületének használatával történő használatához szükséges utasításokat tartalmazza. Ha nem jelenik meg a Catalog Explorer beállítás, használja a megadott Python- vagy SQL-parancsokat.

A kötet képességeinek és használati eseteinek teljes áttekintéséért tekintse meg a Unity Catalog köteteit.

Megjegyzés:

Ez az útmutató MI-funkciókat használ a fájlok elérési út alapján történő feldolgozásához. A Bétában elérhető típus FILE lehetővé teszi, hogy fájlhivatkozásokat és metaadatokat oszlopértékként tároljunk egy táblában. Lásd: FÁJLTÍPUS és strukturálatlan adatok.

Requirements

• Egy Azure Databricks-munkaterület, amelyen engedélyezve van a Unity Catalog.
• CREATE CATALOG jogosultság a metaadat-tárházban. Lásd: Katalógusok létrehozása. Ha nem tud katalógust létrehozni, kérjen hozzáférést a rendszergazdától, vagy használjon olyan meglévő katalógust, amelyben rendelkezik jogosultsággal CREATE SCHEMA .
• Databricks Runtime 14.3 LTS és újabb.
• Mesterséges intelligencia funkciók esetén: Munkaterület egy támogatott régióban.
• Az OpenSharinghez: CREATE SHARE és CREATE RECIPIENT jogosultság a metaadattárra. Lásd: Adatok és AI-eszközök biztonságos megosztása.

1. lépés: Hozzon létre egy kötetet

Hozzon létre egy katalógust, sémát és kötetet a fájlok tárolásához. Részletes kötetkezelési utasításokért lásd : Unity Catalog-kötetek létrehozása és kezelése.

1.1. lépés: Katalógus és séma létrehozása

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

Katalóguskezelő

1. Kattintson Katalógus az oldalsávon.
2. Kattintsona Katalógus létrehozása> gombra.
3. Adja meg unstructured_data_labkatalógusnévként.
4. Kattintson a Létrehozás gombra.
5. Kattintson a Katalógus megtekintése elemre.

A katalógus oldalán:

1. Kattintson a Séma létrehozása gombra.
2. Írja be raw-ként a séma nevét.
3. Kattintson a Létrehozás gombra.

1.2. lépés: Felügyelt kötet létrehozása

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

Katalóguskezelő

A sémaoldalon:

1. Kattintson aKötet> gombra.
2. Adja meg files_volumekötetnévként.
3. Ellenőrizze, hogy a felügyelt kötet ki van-e jelölve.
4. Kattintson a Létrehozás gombra.

2. lépés: Fájlok feltöltése

Fájlok feltöltése a kötetre. Átfogó fájlkezelési példákért lásd: Fájlok használata Unity Catalog-kötetekben.

2.1. lépés: Fájlok feltöltése

Ehhez az oktatóanyaghoz databricks-datasets példákat használhat, vagy feltöltheti saját fájljait a Catalog Explorer felhasználói felületén.

Megjegyzés:

A Python-parancsokkal akkor is másolhat fájlokat a kötetre databricks-datasets , ha nem ismeri a Pythont. A parancsok jegyzetfüzetekben való futtatására vonatkozó utasításokat a Databricks-jegyzetfüzetek kezelése című témakörben találja.

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

Katalóguskezelő

A Python lapon található Python-kód két fájlt (JPG és PDF) tölt fel, valamint egy könyvtárat, amely .txt és .csv fájlokat tartalmaz. Fájlok feltöltése a Catalog Explorerrel:

1. A kötet lapján kattintson a Feltöltés erre a kötetre elemre.
2. A Fájlok feltöltése párbeszédpanel Fájlok csoportjában kattintson a Tallózás gombra, vagy húzza a fájlokat a legördülő zónába.
3. A Célkötet területen ellenőrizze, hogy az előző lépésben létrehozott kötet ki van-e jelölve.

2.2. lépés: A feltöltés ellenőrzése

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

Katalóguskezelő

A fájlok feltöltésekor megjelennek a kötet oldalán. Kattintson egy fájl nevére az előnézet megtekintéséhez, vagy kattintson egy könyvtárra az egyes fájlok megtekintéséhez.

Alternatív megoldás: A %fs magic parancs használata

Használja a %fs magic parancsot:

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

3. lépés: Fájl metaadatainak lekérdezése

Fájlinformációk lekérdezése a kötetben található információk megértéséhez. További lekérdezési mintákért tekintse meg az SQL-vel rendelkező kötetekben lévő fájlok listáját és lekérdezését.

3.1. lépés: Fájl metaadatainak megjelenítése

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)

Katalóguskezelő

A Katalóguskezelő kötetoldalán az egyes fájlok neve (beleértve a bővítményt), a Méret és az Utolsó módosítás dátuma látható.

4. lépés: Fájlok lekérdezése és feldolgozása

Az Azure Databricks AI-függvényekkel tartalmakat nyerhet ki a dokumentumokból és elemezheti a képeket. Az MI funkciók képességeinek teljes áttekintését lásd: Strukturálatlan adatok átalakítása AI függvények segítségével.

Megjegyzés:

Az AI-függvények egy támogatott régióban lévő munkaterületet igényelnek. Lásd : Strukturálatlan adatok átalakítása AI funkciók segítségével.

Ha nem fér hozzá az AI-függvényekhez, használjon inkább standard Python-kódtárakat. Példákért bontsa ki az alábbi alternatív szakaszokat.

4.1. lépés: Dokumentumok elemzése

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)

Alternatív megoldás: PDF-fájlok elemzése AI-függvények nélkül

Ha az AI-függvények nem érhetők el a régióban, használja a Python-kódtárakat:

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

4.2. lépés: Képek elemzése

SQL

SELECT
  path,
  ai_query(
    'system.ai.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(
        'system.ai.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)

Alternatív megoldás: Kép metaadatainak kinyerása AI-függvények nélkül

Kép metaadatainak kinyeréséhez AI-függvények nélkül:

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

4.3. lépés: Szűrés és elemzés fájlnév alapján

Ez a példa azokra a képfájlokra szűr, amelyek fájlnevében szerepel a "rose" karakterlánc.

SQL

SELECT
  path AS file_path,
  ai_query(
    'system.ai.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(
        'system.ai.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)

4.4. lépés: Fájlok csatlakoztatása strukturált táblákkal

Ez a példa sorszámokkal párosítja a fájlokat a taxiutakkal bemutatási célokra. Gyártásban végezzen összekapcsolást jelentős üzleti kulcsokkal.

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)

5. lépés: Hozzáférés-vezérlés alkalmazása

Szabályozhatja, hogy ki tud fájlokat olvasni és írni a kötetekben. A jogosultságok Unity-katalógusban való kezeléséről további információt a Jogosultságok kezelése a Unity Katalógusban című témakörben talál.

5.1. lépés: Hozzáférés biztosítása

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

Katalóguskezelő

1. Lépjen az Engedélyek lapra a kötet oldalán.
2. Kattintson az Engedélyezés gombra.
3. Adja meg egy felhasználó e-mail-címét vagy egy csoport nevét.
4. Válassza ki a megadni kívánt engedélyeket.
5. Kattintson a Megerősítés gombra.

5.2. lépés: Az aktuális jogosultságok megtekintése

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

Katalóguskezelő

A kötetoldal Engedélyek lapján látható, hogy mely felhasználók és csoportok férhetnek hozzá a kötethez.

6. lépés: Növekményes betöltés beállítása

Az Automatikus betöltő használatával automatikusan feldolgozhatja az új fájlokat, amikor azok megérkeznek a kötetbe. Ez a minta hasznos a folyamatos adatbetöltési munkafolyamatokhoz. További betöltési mintákért lásd a gyakori adatbetöltési mintákat.

6.1. lépés: Streamelési tábla létrehozása

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

7. lépés: Fájlok megosztása az OpenSharing használatával

Kötetek biztonságos megosztása más szervezetek felhasználóival az OpenSharing használatával. A megosztás előtt létre kell hoznia egy címzettet. A címzett egy külső szervezetet vagy felhasználót jelöl, aki hozzáférhet a megosztott adatokhoz. A címzettek beállításához tekintse meg az OpenSharing (Databricks-to-Databricks-megosztás) adat címzettjeinek létrehozása című témakört.

7.1. lépés: Megosztás létrehozása és konfigurálása

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

7.2. lépés: Megosztott adatok elérése (címzettként)

SQL

-- View available shares
SHOW SHARES IN PROVIDER <provider_name>;

-- Create a catalog from the share
CREATE CATALOG IF NOT EXISTS shared_documents
USING 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
    USING 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)

8. lépés: Fájlok törlése

Ha már nincs rájuk szükség, távolítsa el a fájlokat.

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)

parancssori felület

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

Alternatív megoldás: Standard Python használata

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

További erőforrások

A kötetek megismerésének folytatása

• Mik azok a Unity Catalog-kötetek?
• Fájlok használata Unity Catalog-kötetekben
• Unity Catalog-kötetek létrehozása és kezelése

A kapcsolódó képességek megismerése

• Tartalomkeresés Unity Catalog kötetekhez
• Strukturálatlan adatok átalakítása AI funkciók segítségével
• Gyakori adatbetöltési minták
• Adatok és AI-eszközök biztonságos megosztása

SQL-függvényhivatkozások

• read_files táblaértékű függvény
• ai_parse_document függvény