處理大量非結構化資料

本頁教你如何使用 Unity 目錄卷來儲存、查詢及處理非結構化資料檔案。 你將學習如何上傳檔案、查詢元資料、利用 AI 功能處理檔案、套用存取控制，以及與其他組織分享磁碟卷。 在可能的情況下，已附上使用目錄檔案總管介面進行此教學的說明。 若未顯示目錄 檔案總管 選項，請使用提供的 Python 或 SQL 指令。

欲全面了解卷的功能與使用案例，請參閱 什麼是 Unity Catalog 卷？。

備註

這個教學使用 AI 函數依路徑處理檔案。 此類型在 Beta FILE 版本中可用，允許你將檔案參考與元資料以欄位值形式儲存在表格中。 請參見 FILE 類型與非結構化資料。

需求

• 一個啟用 Unity Catalog 的 Azure Databricks 工作區。
• CREATE CATALOG 在 Metastore 上享有特權。 請參閱 建立目錄。 如果你無法建立目錄，請向管理員申請存取權限，或使用你有 CREATE SCHEMA 權限的現有目錄。
• Databricks 執行環境 14.3 LTS 及以上版本。
• 針對 AI 功能：支援區域內的工作區域。
• 對於 OpenSharing：在 metastore 上的 CREATE SHARE 和 CREATE RECIPIENT 權限。 請參閱 「安全分享資料與 AI 資產」。

步驟 1：建立一個磁碟區

建立目錄、結構和磁碟區來儲存你的檔案。 如需詳細的卷管理說明，請參閱建立與管理 Unity 目錄卷。

步驟 1.1：建立目錄與架構

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

目錄檢視器

1. 按一下在側邊欄中點擊目錄。
2. 點擊 建立>目錄。
3. 請輸入 unstructured_data_lab 作為 目錄名稱。
4. 點擊 建立。
5. 點擊 檢視目錄。

在目錄頁面上：

1. 點擊 建立架構。
2. 輸入 raw 作為 Schema 名稱。
3. 點擊 建立。

步驟 1.2：建立管理磁碟區

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

目錄檢視器

在結構頁面上：

1. 按一下 建立>磁碟區。
2. 輸入 files_volume 作為 卷名。
3. 確認已選擇 管理磁碟區 。
4. 點擊 建立。

步驟 2：上傳檔案

將檔案上傳到你的雲端硬碟。 如需完整的檔案管理範例，請參閱「 在 Unity 目錄卷中處理檔案」。

步驟 2.1：上傳檔案

你可以用本 databricks-datasets 教學的範例，或用目錄總管的介面上傳自己的檔案。

備註

即使你不熟悉 Python，也可以用 Python 指令將檔案從 databricks-datasets 複製到你的磁碟區。 請參閱 管理 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}")

目錄檢視器

Python 分頁 中的 Python 程式碼會上傳兩個檔案（JPG 和 PDF）以及一個包含 .txt 和 .csv 檔案的目錄。 要使用 Catalog Explorer 上傳檔案：

1. 從卷宗頁點擊上傳至此卷。
2. 在 「上傳檔案 」對話框中，在 「檔案」下，點擊 瀏覽 或拖放檔案到放置區。
3. 在 目標磁碟區，確認你在前一步建立的磁碟區是否被選取。

步驟 2.2：驗證上傳

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

目錄檢視器

上傳檔案後，會出現在磁碟卷頁面。 點擊檔案名稱可預覽，或點擊目錄以查看個別檔案。

另一種選擇：使用 %fs 魔法指令

使用 %fs 魔術指令：

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

步驟 3：查詢檔案元資料

查詢檔案資訊以了解你的卷中內容。 欲了解更多查詢模式，請參閱 使用 SQL 列出及查詢卷中的檔案。

步驟 3.1：顯示檔案元資料

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)

目錄檢視器

目錄總管中的卷頁會顯示每個檔案 的名稱 （包括副檔名）、 大小以及 最後修改 日期。

步驟 4：查詢與處理檔案

使用 Azure Databricks 的 AI 功能從文件中擷取內容並分析圖片。 欲了解 AI 功能能力的完整概述，請參閱 「利用 AI 函數豐富資料」。

備註

AI 函式需要在支援區域內有工作區。 請參見 「利用 AI 函數豐富資料」。

如果你沒有 AI 函式，建議改用標準的 Python 函式庫。 請展開下方的「替代」部分以取得範例。

步驟 4.1：解析文件

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)

替代方案：解析不含 AI 功能的 PDF。

如果你所在地區沒有 AI 函式庫，請使用 Python 函式庫：

%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：分析影像

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)

替代方案：擷取不使用 AI 功能的影像元資料

要提取沒有 AI 功能的影像元資料：

%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：依檔案名稱篩選與分析

此範例篩選包含子字串「rose」的影像檔案。

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)

步驟 4.4：將檔案與結構化資料表合併

此範例使用列號將檔案與計程車行程配對，方便示範。 在生產線上，加入時要有意義的商業鍵。

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：套用存取控制

控制誰能讀寫你卷中的檔案。 欲了解更多關於管理 Unity 目錄權限的資訊，請參閱 「管理 Unity 目錄中的權限」。

步驟 5.1：授權存取權限

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

目錄檢視器

1. 到磁碟區頁面的 權限 標籤。
2. 請按一下 授權。
3. 輸入使用者的電子郵件地址或組名。
4. 選取要授與的許可權。
5. 按一下 [確認]。

步驟 5.2：查看目前的權利

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

目錄檢視器

磁碟區頁面的 權限 標籤顯示哪些使用者和群組有權限存取該磁碟區。

步驟六：設定增量攝取

使用 Auto Loader 自動處理新檔案，當它們進入你的磁碟區時立即處理。 此模式對持續資料擷取工作流程非常有用。 更多資料擷取模式，請參見 常見資料載入模式。

步驟 6.1：建立串流表

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：與OpenSharing分享檔案

使用 OpenSharing 與其他組織的使用者安全分享磁碟區。 你必須先建立收件人才能分享。 收件人代表一個外部組織或使用者，可以存取你的共享資料。 請參閱 建立 OpenSharing 的資料接收者（Databricks-to-Databricks 共享） 以了解接收者設定。

步驟 7.1：建立並設定分享

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：存取共享資料（作為接收者）

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)

步驟八：清理檔案

當檔案不再需要時，請移除它們。

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/

替代方案：使用標準 Python

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

其他資源

繼續學習關於體積的知識

• Unity Catalog 磁碟區是什麼？
• 在 Unity 目錄卷中處理檔案
• 建立與管理 Unity 目錄卷

探索相關功能

• Unity 目錄卷冊的內容搜尋
• 利用 AI 函式豐富資料
• 常見的資料載入模式
• 安全分享資料與 AI 資產

SQL 函式參考

• read_files 數據表值函式
• ai_parse_document函式