Работа с неструктурированными данными в больших объёмах

На этой странице показано, как хранить, запрашивать и обрабатывать неструктурированные файлы данных с помощью томов каталога Unity. Вы узнаете, как отправлять файлы, метаданные запроса, обрабатывать файлы с помощью функций ИИ, применять управление доступом и совместно использовать тома с другими организациями. По возможности были включены инструкции по работе с этим руководством с помощью пользовательского интерфейса обозревателя каталогов. Если параметр обозревателя каталогов не отображается, используйте указанные команды Python или SQL.

Полный обзор возможностей томов и вариантов использования см. в разделе "Что такое тома каталога Unity?".

Замечание

В этом учебном процессе используются функции ИИ для обработки файлов по пути. Доступный в бета-версии, этот FILE тип позволяет хранить ссылки на файлы и метаданные в виде столбцов в таблице. См. тип FILE и неструктурированные данные.

Требования

• Рабочая область Azure Databricks с подключённым Unity Catalog.
• CREATE CATALOG привилегия на хранилище метаданных. См. Создание каталогов. Если вы не можете создать каталог, попросите администратора получить доступ или использовать существующий каталог, где у вас есть привилегии CREATE SCHEMA .
• Databricks Runtime 14.3 LTS и более поздние версии.
• Для функций ИИ: рабочая область в поддерживаемом регионе.
• Для OpenSharing: привилегии CREATE SHARE и CREATE RECIPIENT для хранилища метаданных. Сведения о безопасном использовании данных и ресурсов ИИ см. в статье "Общий доступ к данным и ресурсам ИИ".

Шаг 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;

Питон

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. Введите необработанноеимя схемы.
3. Нажмите кнопку Создать.

Шаг 1.2. Создание управляемого тома

SQL

CREATE VOLUME IF NOT EXISTS files_volume
COMMENT 'Volume for storing unstructured data files';

Питон

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 для копирования файлов из databricks-datasets в ваш том, даже если вы не знакомы с Python. Инструкции по выполнению команд в записных книжках см. в разделе "Управление записными книжками Databricks ".

Питон

# 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 файлы. Чтобы отправить файлы с помощью обозревателя каталогов, выполните следующее:

1. На странице тома нажмите Загрузить в этот том.
2. В диалоговом окне "Отправка файлов" в разделе "Файлы" выберите "Обзор" или перетащите файлы в зону для загрузки.
3. В разделе "Целевой том" убедитесь, что выбран том, созданный на предыдущем шаге.

Шаг 2.2. Проверка отправки

SQL

LIST '/Volumes/unstructured_data_lab/raw/files_volume/';

Питон

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

Питон

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 для извлечения содержимого из документов и анализа изображений. Полный обзор возможностей функций ИИ см. в статье "Обогащение данных с помощью функций ИИ".

Замечание

Для функций ИИ требуется рабочая область в поддерживаемом регионе. См . статью "Обогащение данных с помощью функций ИИ".

Если у вас нет доступа к функциям ИИ, используйте стандартные библиотеки 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'
);

Питон

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)

Альтернатива: анализ PDF-файлов без функций ИИ

Если функции ИИ недоступны в вашем регионе, используйте библиотеки 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;

Питон

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

Шаг 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%';

Питон

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;

Питон

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>`;

Питон

# 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. Нажмите Grant.
3. Введите адрес электронной почты пользователя или имя группы.
4. Выберите разрешения, которые необходимо предоставить.
5. Нажмите кнопку "Подтвердить".

Шаг 5.2. Просмотр текущих привилегий

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Питон

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

Питон

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>;

Питон

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;

Питон

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. Очистка файлов

Удалите файлы, когда они больше не нужны.

Питон

# 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?
• Работа с файлами в томах каталога Unity
• Создание томов каталога Unity и управление ими

Изучение связанных возможностей

• Поиск по содержимому томов Unity Catalog
• Обогащение данных с помощью функций ИИ
• Общие шаблоны загрузки данных
• Безопасное предоставление общего доступа к данным и ресурсам ИИ

Ссылки на функции SQL

• read_files Функция с табличным значением
• ai_parse_document функция