处理大量的非结构化数据

本页介绍如何使用 Unity 目录卷存储、查询和处理非结构化数据文件。你将学习如何上传文件、查询元数据、使用 AI 函数处理文件、应用访问控制，以及与其他组织共享存储卷。在可能的情况下，已包含使用目录浏览器用户界面完成本教程的说明。如果未显示 目录资源管理器 选项，请使用提供的 Python 或 SQL 命令。

有关卷功能和用例的完整概述，请参阅什么是 Unity 目录卷？

注释

本教程使用AI函数按路径处理文件。 Beta版本支持，该 FILE 类型允许你将文件引用和元数据作为表格中的列值存储。参见文件类型和非结构化数据。

Requirements

启用了 Unity 目录的 Azure Databricks 工作区。
CREATE CATALOG 在元存储上的权限。请参阅创建目录。如果无法创建目录，请向管理员申请权限，或使用您拥有 CREATE SCHEMA 权限的现有目录。
Databricks Runtime 14.3 LTS 及更高版本。
对于 AI 函数，需要一个位于受支持区域的工作区。
对于 OpenSharing： 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")

目录浏览器

点击边栏中的目录。
单击“ 创建>目录”。
输入 unstructured_data_lab 作为 目录名称。
单击 “创建” 。
单击“ 查看目录”。

在目录页面上：

单击“ 创建架构”。
输入原始作为 架构名称。
单击 “创建” 。

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

目录浏览器

在架构页上：

单击“ 创建>卷”。
输入 files_volume 为 卷名称。
验证是否选择了 托管卷 。
单击 “创建” 。

步骤 2：上传文件

将文件上传到卷。有关全面的文件管理示例，请参阅在 Unity Catalog 卷中处理文件。

步骤 2.1：上传文件

可以使用本教程中的示例 databricks-datasets ，也可以使用目录资源管理器 UI 上传自己的文件。

注释

即使不熟悉 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 文件的目录。使用目录浏览器上传文件：

在卷页中，单击“ 上传到此卷”。
在“ 上传文件 ”对话框中的“ 文件”下，单击 “浏览 ”或“将文件拖放到放置区域”。
在 “目标卷”下，验证是否选择了在上一步中创建的卷。

步骤 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 magic 命令

%fs使用 magic 命令：

%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 函数，请改用标准 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>`
""")

目录浏览器

转到卷页上的“ 权限 ”选项卡。
单击授权。
输入用户的电子邮件地址或组的名称。
选择要授予的权限。
单击“确认”。

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

目录浏览器

卷页上的“ 权限 ”选项卡显示哪些用户和组有权访问该卷。

步骤 6：设置增量引入

使用自动加载程序在到达卷时自动处理新文件。此模式适用于连续数据引入工作流。有关更多引入模式，请参阅常见数据加载模式。

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

使用 OpenSharing 与其他组织中的用户安全地共享卷。在共享之前，必须创建收件人。收件人表示可以访问共享数据的外部组织或用户。有关收件人设置，请参阅创建 OpenSharing 的数据收件人（Databricks 到 Databricks 共享）。

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)

步骤 8：清理文件

不再需要文件时删除文件。

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

其他资源

继续了解卷

SQL 函数引用

Last updated on 2026-09-11

处理大量的非结构化数据

Requirements

步骤 1：创建卷

步骤 1.1：创建目录和架构

SQL

Python

目录浏览器

步骤 1.2：创建托管卷

SQL

Python

目录浏览器

步骤 2：上传文件

步骤 2.1：上传文件

Python

目录浏览器

步骤 2.2：验证上传

SQL

Python

目录浏览器

步骤 3：查询文件元数据

步骤 3.1：显示文件元数据

SQL

Python

目录浏览器

步骤 4：查询和处理文件

步骤 4.1：分析文档

SQL

Python

步骤 4.2：分析图像

SQL

Python

步骤 4.3：按文件名筛选和分析

SQL

Python

步骤 4.4：将文件与结构化表联接

SQL

Python

步骤 5：应用访问控制

步骤 5.1：授予访问权限

SQL

Python

目录浏览器

步骤 5.2：查看当前权限

SQL

Python

目录浏览器

步骤 6：设置增量引入

步骤 6.1：创建流式处理表

SQL

Python

步骤 7：使用 OpenSharing 共享文件

步骤 7.1：创建和配置共享

SQL

Python

步骤 7.2：访问共享数据（作为收件人）

SQL

Python

步骤 8：清理文件

Python

CLI

其他资源

继续了解卷

探索相关功能

SQL 函数引用

其他资源