볼륨에서 구조화되지 않은 데이터 작업

이 페이지에서는 Unity 카탈로그 볼륨을 사용하여 구조화되지 않은 데이터 파일을 저장, 쿼리 및 처리하는 방법을 보여 줍니다. 파일을 업로드하고, 메타데이터를 쿼리하고, AI 함수를 사용하여 파일을 처리하고, 액세스 제어를 적용하고, 볼륨을 다른 조직과 공유하는 방법을 알아봅니다. 가능한 경우 카탈로그 탐색기 UI를 사용하여 이 자습서를 진행하기 위한 지침이 포함되어 있습니다. 카탈로그 탐색기 옵션이 표시되지 않으면 제공된 Python 또는 SQL 명령을 사용합니다.

볼륨 기능 및 사용 사례에 대한 전체 개요는 Unity 카탈로그 볼륨이란?을 참조하세요.

요구 사항

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;

파이썬

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

파이썬

spark.sql("""
    CREATE VOLUME IF NOT EXISTS files_volume
    COMMENT 'Volume for storing unstructured data files'
""")

카탈로그 탐색기

스키마 페이지에서 다음을 수행합니다.

만들기>볼륨을 클릭합니다.
볼륨 이름으로files_volume 입력합니다.
관리되는 볼륨 이 선택되어 있는지 확인합니다.
만들기를 클릭합니다.

2단계: 파일 업로드

볼륨에 파일을 업로드하십시오. 포괄적인 파일 관리 예제는 Unity 카탈로그 볼륨의 파일 작업을 참조하세요.

2.1단계: 파일 업로드

이 자습서의 databricks-datasets 예제를 사용하거나 카탈로그 탐색기 UI를 사용하여 사용자 고유의 파일을 업로드할 수 있습니다.

비고

Python에 익숙하지 않은 경우에도 Python 명령을 사용하여 볼륨으로 databricks-datasets 파일을 복사할 수 있습니다. Notebook에서 명령을 실행하는 방법에 대한 지침 은 Databricks Notebook 관리를 참조하세요.

파이썬

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

파이썬

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 AI 함수를 사용하여 문서에서 콘텐츠를 추출하고 이미지를 분석합니다. AI 함수 기능에 대한 전체 개요는 AI Functions를 사용하여 데이터 보강을 참조하세요.

비고

AI 함수에는 지원되는 지역의 작업 영역이 필요합니다. AI Functions를 사용하여 데이터 보강을 참조하세요.

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

파이썬

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;

파이썬

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

파이썬

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

카탈로그 탐색기

볼륨 페이지의 사용 권한 탭으로 이동합니다.
Grant를 클릭합니다.
사용자의 이메일 주소 또는 그룹 이름을 입력합니다.
부여할 권한을 선택합니다.
확인을 클릭합니다.

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

OpenSharing을 사용하여 다른 조직의 사용자와 볼륨을 안전하게 공유합니다. 공유하기 전에 받는 사람을 만들어야 합니다. 받는 사람은 공유 데이터에 액세스할 수 있는 외부 조직 또는 사용자를 나타냅니다. 수신자 설정은 OpenSharing용 데이터 수신자 만들기(Databricks-to-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>;

파이썬

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

볼륨에서 구조화되지 않은 데이터 작업

요구 사항

1단계: 볼륨 만들기

1.1단계: 카탈로그 및 스키마 만들기

SQL

파이썬

카탈로그 탐색기

1.2단계: 관리되는 볼륨 만들기

SQL

파이썬

카탈로그 탐색기

2단계: 파일 업로드

2.1단계: 파일 업로드

파이썬

카탈로그 탐색기

2.2단계: 업로드 확인

SQL

파이썬

카탈로그 탐색기

3단계: 파일 메타데이터 쿼리

3.1단계: 파일 메타데이터 표시

SQL

파이썬

카탈로그 탐색기

4단계: 파일 쿼리 및 처리

4.1단계: 문서 구문 분석

SQL

파이썬

4.2단계: 이미지 분석

SQL

파이썬

4.3단계: 파일 이름으로 필터링 및 분석

SQL

파이썬

4.4단계: 구조화된 테이블을 사용하여 파일 조인

SQL

파이썬

5단계: 액세스 제어 적용

5.1단계: 액세스 권한 부여

SQL

파이썬

카탈로그 탐색기

5.2단계: 현재 권한 보기

SQL

파이썬

카탈로그 탐색기

6단계: 증분 수집 설정

6.1단계: 스트리밍 테이블 만들기

SQL

파이썬

7단계: OpenSharing을 사용하여 파일 공유

7.1단계: 공유 만들기 및 구성

SQL

파이썬

7.2단계: 공유 데이터 액세스(받는 사람)

SQL

파이썬

8단계: 파일 정리

파이썬

CLI

추가 리소스

볼륨에 대한 학습을 계속합니다.

관련 기능 살펴보기

SQL 함수 참조

피드백

추가 리소스