Bekerja dengan data yang tidak terstruktur dalam jumlah besar

Halaman ini memperlihatkan kepada Anda cara menyimpan, mengkueri, dan memproses file data yang tidak terstruktur menggunakan volume Katalog Unity. Anda akan mempelajari cara mengunggah file, metadata kueri, memproses file dengan fungsi AI, menerapkan kontrol akses, dan berbagi volume dengan organisasi lain. Jika memungkinkan, instruksi untuk bekerja melalui tutorial ini menggunakan UI Catalog Explorer telah disertakan. Jika tidak ada opsi Catalog Explorer yang ditampilkan, gunakan perintah Python atau SQL yang disediakan.

Untuk gambaran umum lengkap kemampuan volume dan kasus penggunaan, lihat Apa itu volume Katalog Unity?.

Nota

Tutorial ini menggunakan fungsi AI untuk memproses file berdasarkan jalur. Tersedia di Beta, tipe ini FILE memungkinkan Anda menyimpan referensi file dan metadata sebagai nilai kolom dalam sebuah tabel. Lihat tipe FILE dan data tidak terstruktur.

Persyaratan

• Ruang kerja Azure Databricks dengan Unity Catalog diaktifkan.
• CREATE CATALOG hak istimewa di metastore. Lihat "Buat katalog". Jika Anda tidak dapat membuat katalog, minta admin Anda untuk mengakses atau menggunakan katalog yang sudah ada di mana Anda memiliki CREATE SCHEMA hak istimewa.
• Databricks Runtime 14.3 LTS ke atas.
• Untuk fungsi AI: Ruang kerja di wilayah yang didukung.
• Untuk OpenSharing: hak akses CREATE SHARE dan CREATE RECIPIENT pada metastore. Lihat Berbagi data dan aset AI dengan aman.

Langkah 1: Membuat volume

Buat katalog, skema, dan volume untuk menyimpan file Anda. Untuk instruksi manajemen volume terperinci, lihat Membuat dan mengelola volume Katalog Unity.

Langkah 1.1: Membuat katalog dan skema

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;

Phyton

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

Eksplorer Katalog

1. Klik Katalog di bilah samping.
2. Klik Buat>katalog.
3. Masukkan unstructured_data_lab sebagai Nama katalog.
4. Klik Buat.
5. Klik Tampilkan katalog.

Pada halaman katalog:

1. Klik Buat skema.
2. Masukkan raw sebagai Schema Name.
3. Klik Buat.

Langkah 1.2: Membuat volume terkelola

SQL

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

Phyton

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

Eksplorer Katalog

Pada halaman skema:

1. Klik Buat>Volume.
2. Masukkan files_volume sebagai nama Volume.
3. Pastikan Volume terkelola telah dipilih.
4. Klik Buat.

Langkah 2: Unggah file

Unggah file ke volume Anda. Untuk contoh manajemen file yang komprehensif, lihat Bekerja dengan file dalam volume Katalog Unity.

Langkah 2.1: Mengunggah file

Anda dapat menggunakan contoh dari databricks-datasets untuk tutorial ini, atau mengunggah file Anda sendiri menggunakan UI Catalog Explorer.

Nota

Anda dapat menggunakan perintah Python untuk menyalin file dari databricks-datasets ke volume Anda meskipun Anda tidak terbiasa dengan Python. Lihat Mengelola buku catatan Databricks untuk instruksi tentang menjalankan perintah di buku catatan.

Phyton

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

Eksplorer Katalog

Kode Python di tab Python mengunggah dua file (JPG dan PDF) dan direktori yang menyertakan .txt file dan .csv . Untuk mengunggah file menggunakan Catalog Explorer:

1. Dari halaman volume, klik Unggah ke volume ini.
2. Dalam dialog Unggah file , di bawah File, klik telusuri atau seret dan letakkan file ke zona drop.
3. Pada bagian Volume tujuan, pastikan bahwa volume yang Anda buat di langkah sebelumnya telah dipilih.

Langkah 2.2: Verifikasi unggahan

SQL

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

Phyton

files = dbutils.fs.ls("/Volumes/unstructured_data_lab/raw/files_volume/")
for f in files:
    print(f"{f.name}\t{f.size} bytes")

Eksplorer Katalog

Saat file diunggah, file akan muncul di halaman volume. Klik nama file untuk melihat pratinjau, atau klik direktori untuk melihat file individual.

Alternatif: Gunakan perintah magic %fs

%fs Gunakan perintah ajaib:

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

Langkah 3: Mengkueri metadata file

Periksa informasi berkas untuk memahami apa yang ada dalam volume Anda. Untuk pola kueri lainnya, lihat Mencantumkan dan mengkueri file dalam volume dengan SQL.

Langkah 3.1: Tampilkan metadata file

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

Phyton

df = (
    spark.read
    .format("binaryFile")
    .option("recursiveFileLookup", "true")
    .load("/Volumes/unstructured_data_lab/raw/files_volume/")
)

df.select("path", "modificationTime", "length").show(truncate=False)

Eksplorer Katalog

Halaman volume di Catalog Explorer memperlihatkan Nama setiap file (termasuk ekstensi), Ukuran, dan Tanggal terakhir diubah .

Langkah 4: Mengkueri dan memproses file

Gunakan fungsi Azure Databricks AI untuk mengekstrak konten dari dokumen dan menganalisis gambar. Untuk gambaran lengkap tentang kemampuan fungsi AI, lihat Transformasi data tak terstruktur menggunakan Fungsi AI.

Nota

Fungsi AI memerlukan ruang kerja di wilayah yang didukung. Lihat Transformasi data tidak terstruktur menggunakan Fungsi AI.

Jika Anda tidak memiliki akses ke fungsi AI, gunakan pustaka Python standar sebagai gantinya. Perluas bagian Alternatif di bawah ini untuk contoh.

Langkah 4.1: Mengurai dokumen

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

Phyton

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)

Alternatif: Mengurai PDF tanpa fungsi AI

Jika fungsi AI tidak tersedia di wilayah Anda, gunakan pustaka 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"))

Langkah 4.2: Menganalisis gambar

SQL

SELECT
  path,
  ai_query(
    'system.ai.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;

Phyton

result_df = spark.sql("""
    SELECT
      path,
      ai_query(
        'system.ai.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)

Alternatif: Mengekstrak metadata gambar tanpa fungsi AI

Untuk mengekstrak metadata gambar tanpa fungsi 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.*"))

Langkah 4.3: Memfilter dan menganalisis berdasarkan nama file

Contoh ini memfilter file gambar dengan substring "rose" dalam nama file mereka.

SQL

SELECT
  path AS file_path,
  ai_query(
    'system.ai.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%';

Phyton

result_df = spark.sql("""
    SELECT
      path AS file_path,
      ai_query(
        'system.ai.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)

Langkah 4.4: Menggabungkan file dengan tabel terstruktur

Contoh ini menggunakan nomor baris untuk memasangkan file dengan perjalanan taksi untuk tujuan demonstrasi. Dalam produksi, bergabunglah dengan kunci bisnis yang bermakna.

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;

Phyton

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)

Langkah 5: Menerapkan kontrol akses

Mengontrol siapa yang dapat membaca dan menulis file dalam volume Anda. Untuk mempelajari selengkapnya tentang mengelola hak istimewa di Unity Catalog, lihat Mengelola hak istimewa di Unity Catalog.

Langkah 5.1: Memberikan akses

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

Phyton

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

Eksplorer Katalog

1. Buka tab Izin di halaman volume.
2. Klik pada Berikan.
3. Masukkan alamat email untuk pengguna atau nama grup.
4. Pilih izin yang akan diberikan.
5. Klik tombol Konfirmasi.

Langkah 5.2: Melihat hak istimewa saat ini

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Phyton

display(spark.sql("SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume"))

Eksplorer Katalog

Tab Izin pada halaman volume memperlihatkan pengguna dan grup mana yang memiliki akses ke volume.

Langkah 6: Menyiapkan penyerapan bertahap

Gunakan Auto Loader untuk memproses file baru secara otomatis saat file tersebut tiba di volume Anda. Pola ini berguna untuk alur kerja penyerapan data berkelanjutan. Untuk pola penyerapan lainnya, lihat Pola pemuatan data umum.

Langkah 6.1: Membuat tabel streaming

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

Phyton

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

Langkah 7: Berbagi file dengan OpenSharing

Bagikan volume dengan aman dengan pengguna di organisasi lain menggunakan OpenSharing. Anda harus membuat penerima sebelum berbagi. Penerima mewakili organisasi eksternal atau pengguna yang dapat mengakses data bersama Anda. Lihat Pembuatan penerima data untuk OpenSharing (berbagi data Databricks-ke-Databricks) untuk penyiapan penerima.

Langkah 7.1: Membuat dan mengonfigurasi berbagi

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

Phyton

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

Langkah 7.2: Mengakses data bersama (sebagai penerima)

SQL

-- View available shares
SHOW SHARES IN PROVIDER <provider_name>;

-- Create a catalog from the share
CREATE CATALOG IF NOT EXISTS shared_documents
USING 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;

Phyton

spark.sql("SHOW SHARES IN PROVIDER <provider_name>").show()

spark.sql("""
    CREATE CATALOG IF NOT EXISTS shared_documents
    USING 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)

Langkah 8: Bersihkan file

Hapus file saat file tidak lagi diperlukan.

Phyton

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

Alternatif: Gunakan Python standar

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

Sumber daya tambahan

Lanjutkan mempelajari tentang volume

• Apa yang dimaksud dengan volume pada Katalog Unity?
• Bekerja dengan file-file pada volume Katalog Unity
• Membuat dan mengelola volume Katalog Unity

Menjelajahi kemampuan terkait

• Pencarian konten untuk volume Unity Catalog
• Transformasi data tidak terstruktur menggunakan Fungsi AI
• Pola pemuatan data umum
• Berbagi data dan aset AI dengan aman

Referensi fungsi SQL

• read_files fungsi nilai tabel
• ai_parse_document fungsi