Unity Catalog Volumes 심층 분석: 비정형 데이터 관리

/Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

dbfs:/Volumes/<catalog>/<schema>/<volume>/<path>/<file-name>

CREATE VOLUME IF NOT EXISTS my_catalog.my_schema.my_volume
COMMENT 'Managed volume for unstructured data files';

spark.sql("""
    CREATE VOLUME IF NOT EXISTS my_catalog.my_schema.my_volume
    COMMENT 'Managed volume for unstructured data files'
""")

CREATE EXTERNAL VOLUME my_catalog.my_schema.my_external_volume
LOCATION 's3://my-bucket/my-prefix/volumes/'
COMMENT 'External volume pointing to existing S3 storage';

# 파일 업로드
dbutils.fs.cp(
    "dbfs:/databricks-datasets/flower_photos/roses/10090824183_d02c613f10_m.jpg",
    "/Volumes/my_catalog/raw/files_volume/rose.jpg"
)

# 파일 목록 확인
files = dbutils.fs.ls("/Volumes/my_catalog/raw/files_volume/")
for f in files:
    print(f"{f.name}\t{f.size} bytes")

# 파일 삭제
dbutils.fs.rm("/Volumes/my_catalog/raw/files_volume/old_file.jpg")

# 디렉토리 재귀 삭제
dbutils.fs.rm("/Volumes/my_catalog/raw/files_volume/old_dir/", recurse=True)

-- 파일 목록 확인
LIST '/Volumes/my_catalog/raw/files_volume/';

-- 파일 메타데이터 조회
SELECT
  path,
  _metadata.file_name,
  _metadata.file_size,
  _metadata.file_modification_time
FROM read_files(
  '/Volumes/my_catalog/raw/files_volume/',
  format => 'binaryFile'
);

# 파일 목록
databricks fs ls dbfs:/Volumes/my_catalog/raw/files_volume/

# 파일 삭제
databricks fs rm dbfs:/Volumes/my_catalog/raw/files_volume/old_file.jpg

# 디렉토리 재귀 삭제
databricks fs rm -r dbfs:/Volumes/my_catalog/raw/files_volume/old_dir/

# 텍스트 파일 쓰기
with open("/Volumes/my_catalog/raw/files_volume/hello.txt", "w") as f:
    f.write("Hello, Volumes!")

# 텍스트 파일 읽기
with open("/Volumes/my_catalog/raw/files_volume/hello.txt", "r") as f:
    content = f.read()
    print(content)

# 이진 파일 읽기 (예: 이미지)
import os
os.remove("/Volumes/my_catalog/raw/files_volume/old_file.jpg")

SELECT
  path AS file_path,
  ai_parse_document(content, map('version', '2.0')) AS parsed_content
FROM read_files(
  '/Volumes/my_catalog/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.pdf'
);

%pip install PyPDF2

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/my_catalog/raw/files_volume/")
result_df = df.withColumn("text_content", extract_pdf_text("content"))
display(result_df.select("path", "text_content"))

SELECT
  path,
  ai_query(
    'databricks-llama-4-maverick',
    'Describe this image in one sentence:',
    files => content
  ) AS description
FROM read_files(
  '/Volumes/my_catalog/raw/files_volume/',
  format => 'binaryFile',
  fileNamePattern => '*.{jpg,jpeg,png}'
)
WHERE _metadata.file_size < 5000000;

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/my_catalog/raw/files_volume/',
    format => 'binaryFile'
  )
)
SELECT
  f.path,
  f.file_name,
  f.length
FROM files_with_row f
LIMIT 10;

from pyspark.sql.functions import current_timestamp, col

dbutils.fs.mkdirs("/Volumes/my_catalog/raw/files_volume/incoming/")

df = spark.readStream.format("cloudFiles") \
    .option("cloudFiles.format", "binaryFile") \
    .option("pathGlobFilter", "*.pdf") \
    .load("/Volumes/my_catalog/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/my_catalog/raw/files_volume/_checkpoints/docs") \
    .trigger(availableNow=True) \
    .toTable("my_catalog.processed.document_ingestion")

query.awaitTermination()

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/my_catalog/raw/files_volume/incoming/',
  format => 'binaryFile'
));

-- 읽기 권한 부여
GRANT READ VOLUME ON VOLUME my_catalog.raw.files_volume
TO `data-science-team`;

-- 읽기 및 쓰기 권한 부여
GRANT READ VOLUME, WRITE VOLUME ON VOLUME my_catalog.raw.files_volume
TO `data-engineering-team`;

-- 모든 권한 부여
GRANT ALL PRIVILEGES ON VOLUME my_catalog.raw.files_volume
TO `admin-group`;

-- 현재 부여된 권한 조회
SHOW GRANTS ON VOLUME my_catalog.raw.files_volume;

spark.sql("""
    GRANT READ VOLUME ON VOLUME my_catalog.raw.files_volume
    TO `data-science-team`
""")

-- Share 생성
CREATE SHARE IF NOT EXISTS unstructured_data_share
COMMENT 'Document files for partner organizations';

-- Volume 추가
ALTER SHARE unstructured_data_share
ADD VOLUME my_catalog.raw.files_volume;

-- Recipient 생성
CREATE RECIPIENT IF NOT EXISTS partner_org
USING ID '<recipient-sharing-identifier>';

-- 액세스 권한 부여
GRANT SELECT ON SHARE unstructured_data_share
TO RECIPIENT partner_org;

-- 사용 가능한 Share 조회
SHOW SHARES IN PROVIDER <provider_name>;

-- Share로부터 카탈로그 생성
CREATE CATALOG IF NOT EXISTS shared_documents
FROM SHARE <provider_name>.unstructured_data_share;

-- 공유된 파일 쿼리
SELECT * EXCEPT (content), _metadata
FROM read_files(
  '/Volumes/shared_documents/raw/files_volume/',
  format => 'binaryFile'
)
LIMIT 10;