Geniş büyüklüklerdeki yapılandırılmamış verilerle çalışmak

Bu sayfada Unity Kataloğu birimlerini kullanarak yapılandırılmamış veri dosyalarını depolama, sorgulama ve işleme işlemleri gösterilir. Dosyaları karşıya yüklemeyi, meta verileri sorgulamayı, yapay zeka işlevleriyle dosyaları işlemeyi, erişim denetimi uygulamayı ve birimleri diğer kuruluşlarla paylaşmayı öğreneceksiniz. Mümkün olduğunda, Katalog Gezgini kullanıcı arabirimini kullanarak bu öğreticide çalışmaya yönelik yönergeler eklenmiştir. Katalog Gezgini seçeneği gösterilmiyorsa, sağlanan Python veya SQL komutlarını kullanın.

Birim özelliklerine ve kullanım örneklerine tam bir genel bakış için bkz. Unity Kataloğu birimleri nedir?.

Gereksinimler

• Unity Kataloğu'nu etkinleştirmiş bir Azure Databricks çalışma alanı.
• CREATE CATALOG meta veri deposunda ayrıcalık. Bkz. Kataloglar oluştur. Katalog oluşturamıyorsanız yöneticinizden erişim isteyin veya ayrıcalığınız CREATE SCHEMA olan mevcut bir kataloğu kullanın.
• Databricks Runtime 14.3 LTS ve üzeri.
• Yapay zeka işlevleri için: Desteklenen bir bölgedeki çalışma alanı.
• OpenSharing için: meta veri deposu üzerindeki CREATE SHARE ve CREATE RECIPIENT ayrıcalıkları. Bkz. Verileri ve yapay zeka varlıklarını güvenli bir şekilde paylaşma.

1. Adım: Birim oluştur

Dosyalarınızı depolamak için bir katalog, şema ve birim oluşturun. Ayrıntılı birim yönetimi yönergeleri için bkz. Unity Kataloğu birimlerini oluşturma ve yönetme.

1.1. Adım: Katalog ve şema oluşturma

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;

Piton

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

Katalog Tarayıcısı

1. Kenar çubuğunda katalog.
2. Katalog oluştur'a> tıklayın.
3. Katalog adı olarak unstructured_data_lab girin.
4. Oluştur'utıklayın.
5. Kataloğu görüntüle'ye tıklayın.

Katalog sayfasında:

1. Şema oluştur'a tıklayın.
2. Şema adı olarak ham değerini girin.
3. Oluştur'utıklayın.

1.2. Adım: Yönetilen birim oluşturma

SQL

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

Piton

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

Katalog Tarayıcısı

Şema sayfasında:

1. Birim> tıklayın.
2. Birim adı olarak files_volume girin.
3. Yönetilen birimin seçili olduğunu doğrulayın.
4. Oluştur'utıklayın.

2. Adım: Dosyaları karşıya yükleme

Dosyaları biriminize yükleyin. Kapsamlı dosya yönetimi örnekleri için bkz. Unity Kataloğu birimlerindeki dosyalarla çalışma.

2.1. Adım: Dosyaları karşıya yükleme

Bu öğretici için databricks-datasets örnekleri kullanabilir veya Katalog Gezgini arayüzünü kullanarak kendi dosyalarınızı yükleyebilirsiniz.

Piton

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

Katalog Tarayıcısı

Python sekmesindeki Python kodu iki dosyayı (JPG ve PDF) ve .txt dosyaları içeren bir .csv dizinini karşıya yükler. Katalog Gezgini'ne kullanarak dosya yüklemek için:

1. Birim sayfasında Bu birime yükle'ye tıklayın.
2. Dosyaları karşıya yükle iletişim kutusundaki Dosyalar'ın altında dosyalara göz at'a tıklayın veya dosyaları bırakma bölgesine sürükleyip bırakın.
3. Hedef birim'in altında, önceki adımda oluşturduğunuz birimin seçili olduğunu doğrulayın.

2.2. Adım: Karşıya yüklemeyi doğrulama

SQL

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

Piton

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

Katalog Tarayıcısı

Dosyalar karşıya yüklendiğinde disk bölümü sayfasında görünür. Önizlemeyi görmek için bir dosya adına tıklayın veya tek tek dosyaları görüntülemek için bir dizine tıklayın.

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

3. Adım: Dosya meta verilerini sorgulama

Hacminizde ne olduğunu anlamak için dosya bilgilerini sorgulayın. Daha fazla sorgu deseni için bkz. SQL ile birimlerdeki dosyaları listeleme ve sorgulama.

3.1. Adım: Dosya meta verilerini gösterme

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

Piton

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

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

Katalog Tarayıcısı

Katalog Gezgini'ndeki birim sayfasında her dosyanın Adı (uzantı dahil), Boyut ve Son değiştirme tarihi gösterilir.

4. Adım: Dosyaları sorgulama ve işleme

Belgelerden içerik ayıklamak ve görüntüleri analiz etmek için Azure Databricks AI işlevlerini kullanın. Yapay zeka işlevi özelliklerine tam bir genel bakış için bkz. Yapay zeka İşlevlerini kullanarak verileri zenginleştirme.

4.1. Adım: Belgeleri ayrıştırma

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

Piton

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)

%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. Adım: Görüntüleri analiz etme

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;

Piton

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. Adım: Dosya adına göre filtreleme ve analiz etme

Bu örnek, dosya adında "rose" alt dizesini içeren görüntü dosyalarına filtre ekler.

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

Piton

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. Adım: Yapılandırılmış tablolarla dosyaları birleştirme

Bu örnek, tanıtım amacıyla dosyaları taksi gezileriyle eşleştirmek için satır numaralarını kullanır. Üretimde anlamlı iş anahtarlarına katılın.

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;

Piton

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. Adım: Erişim denetimi uygulama

Birimlerinizdeki dosyaları kimlerin okuyup yazabileceğini denetleyin. Unity Kataloğu'nda ayrıcalıkları yönetme hakkında daha fazla bilgi edinmek için bkz. Unity Kataloğu'nda ayrıcalıkları yönetme.

5.1. Adım: Erişim izni verme

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

Piton

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

Katalog Tarayıcısı

1. Birim sayfasındaki İzinler sekmesine gidin.
2. İzin Ver'e tıklayın.
3. Bir kullanıcının e-posta adresini veya grubun adını girin.
4. Vermek istediğiniz izinleri seçin.
5. Onayla'yı tıklatın.

5.2. Adım: Geçerli ayrıcalıkları görüntüleme

SQL

SHOW GRANTS ON VOLUME unstructured_data_lab.raw.files_volume;

Piton

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

Katalog Tarayıcısı

Birim sayfasındaki İzinler sekmesi, hangi kullanıcıların ve grupların birime erişimi olduğunu gösterir.

6. Adım: Artımlı alımı ayarlama

Biriminize gelen yeni dosyaları otomatik olarak işlemek için Otomatik Yükleyici'yi kullanın. Bu desen, sürekli veri alımı iş akışları için kullanışlıdır. Daha fazla veri alma deseni için bkz. Yaygın veri yükleme desenleri.

6.1. Adım: Akış tablosu oluşturma

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

Piton

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. Adım: OpenSharing ile dosya paylaşma

OpenSharing kullanarak birimleri diğer kuruluşlardaki kullanıcılarla güvenli bir şekilde paylaşın. Paylaşmadan önce bir alıcı oluşturmanız gerekir. Alıcı, paylaşılan verilerinize erişebilen bir dış kuruluşu veya kullanıcıyı temsil eder. Alıcı kurulumu için OpenSharing için veri alıcıları oluşturma (Databricks’ten Databricks’e paylaşım) bölümüne bakın.

7.1. Adım: Paylaşım oluşturma ve yapılandırma

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

Piton

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. Adım: Paylaşılan verilere erişme (alıcı olarak)

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;

Piton

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. Adım: Dosyaları temizleme

Artık gerekli olmayan dosyaları kaldırın.

Piton

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

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

Ek kaynaklar

Birimler hakkında öğrenmeye devam edin

• Unity Kataloğu birimleri nelerdir?
• Unity Kataloğu birimlerindeki dosyalarla çalışma
• Unity Kataloğu birimleri oluşturma ve yönetme

İlgili özellikleri keşfetme

• Yapay zeka işlevlerini kullanarak verileri zenginleştirme
• Yaygın veri yükleme desenleri
• Verileri ve yapay zeka varlıklarını güvenli bir şekilde paylaşma

SQL işlevi başvuruları

• read_files tablo değerli fonksiyon
• ai_parse_document işlev