Tutorial: Membangun pipeline pemrosesan file dengan tipe FILE

Important

Fitur ini ada di Beta. Untuk menggunakannya, admin workspace harus mengaktifkan Tipe File dari halaman Pratinjau . Lihat Kelola Pratinjau Azure Databricks.

Dalam tutorial ini, Anda membangun sistem medallion dengan pipeline Lakeflow yang memproses dokumen tidak terstruktur secara end-to-end. Contoh ini menggunakan set data sampel samples.sec.contracts, yaitu kumpulan perjanjian hukum yang diajukan ke SEC dan disimpan dalam format PDF di volume Unity Catalog.

Pipeline mengolah PDF sebagai referensi terkelola FILE dengan Auto Loader, mengurai setiap dokumen dengan fungsi AI, mengklasifikasikannya ke dalam jenis perjanjian, dan mengekstrak field terstruktur untuk setiap tipe.

Untuk referensi tipe, lihat FILE tipe.

Dalam tutorial ini, Anda akan:

Ingest PDF kontrak secara inkremental dari volume sebagai referensi terkelola FILE dengan Auto Loader.
Urai setiap dokumen dengan ai_parse_document fungsi dan klasifikasikannya dengan ai_classify fungsi.
Ekstrak bidang terstruktur untuk setiap jenis perjanjian menggunakan ai_extractfungsi.

Hasilnya adalah pipeline bergaya medali: perunggu (referensi terkelola FILE mentah), perak (dokumen yang telah dianalisis dan diklasifikasikan), dan emas (field yang diekstrak berdasarkan jenis perjanjian). Lihat Apa arsitektur medali lakehouse? untuk informasi selengkapnya. Lapisan perunggu adalah tabel streaming yang secara bertahap menyerap file, dan lapisan perak serta emas adalah tampilan termaterialisasi yang dihitung ulang hanya saat input berubah.

Persyaratan

Untuk menyelesaikan tutorial ini, Anda harus memenuhi persyaratan berikut:

Masuk ke ruang kerja Azure Databricks dengan Unity Catalog diaktifkan.
Pastikan tipe FILE diaktifkan untuk ruang kerja Anda. Admin ruang kerja dapat mengaktifkannya dari halaman Pratinjau. Lihat Kelola Pratinjau Azure Databricks.
Memiliki izin untuk membuat tabel dalam skema dan membuat pipeline.
Miliki volume Unity Catalog yang dapat Anda tulisi. Anda mendeklarasikan volume ini sebagai penyimpanan terkelola tabel perungguFileSpace, dan Unity Catalog menyalin file yang dimasukkan ke dalamnya sebagai penyimpanan terkelola.
Gunakan saluran Pratinjau.

samples.sec.contracts Dataset tersedia secara default di semua workspace. Tutorial ini menyimpan PDF yang dimasukkan sebagai FILE MANAGED referensi: Unity Catalog menyalin setiap file ke volume yang Anda deklarasikan sebagai milik tabel FileSpace dan mengelolanya bersama tabel, sehingga menghapus baris membuat file yang direferensikan memenuhi syarat untuk pengumpulan sampah dan tabel beserta file-filenya tetap sinkron. Untuk menyesuaikan pipeline ke PDF Anda sendiri, arahkan jalur sumber ke volume yang berisi file Anda. Untuk opsi ingestion lainnya, lihat Ingest file sebagai tipe FILE.

Buat pipeline pemrosesan file

Alur pemrosesan memproses dokumen dalam tiga tahap.

Langkah 1. Perunggu: mengimpor PDF mentah sebagai referensi FILE yang dikelola

Gunakan Auto Loader untuk membaca berkas PDF kontrak secara bertahap dari volume penyimpanan. Membaca berkas dengan format => 'file' mengambil referensi dan metadata untuk setiap berkas tanpa mematerialisasikan byte-nya. Mendeklarasikan kolom sebagai FILE MANAGED akan menyalin setiap file ke FileSpace tabel, volume yang Anda atur dengan properti tabel databricks.filespace-preview, sehingga Unity Catalog mengelola file dengan tabel tersebut.

SQL

CREATE OR REFRESH STREAMING TABLE raw_contracts (
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE MANAGED
)
TBLPROPERTIES ('databricks.filespace-preview' = '/Volumes/my_catalog/my_schema/filespace/')
AS SELECT *
  FROM STREAM read_files(
    '/Volumes/samples/sec/contracts/',
    format => 'file');

Python

from pyspark import pipelines as dp

@dp.table(
  name="raw_contracts",
  schema="path STRING, size BIGINT, modification_time TIMESTAMP, file FILE MANAGED",
  table_properties={"databricks.filespace-preview": "/Volumes/my_catalog/my_schema/filespace/"}
)
def raw_contracts():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .load("/Volumes/samples/sec/contracts/")
  )

Berfungsi untuk file besar: PDF besar berada di FileSpace tabel, sementara baris tabel hanya menyimpan referensi ringan FILE (uri, size, content_type, checksum). Bandingkan hal ini dengan tipe BINARY, yang menyisipkan byte secara langsung ke dalam baris.
Siklus hidup file yang dikelola: Unity Catalog menyalin setiap file yang dimasukkan ke dalam FileSpace milik tabel dan mengelolanya bersama tabel: menghapus baris membuat file yang direferensikan memenuhi syarat untuk pengumpulan sampah, sehingga tabel dan file-filenya tetap sinkron. Untuk detailnya, lihat FILE MANAGED dan FILE EXTERNAL.
Pemrosesan inkremental: tabel streaming secara bertahap menghisap file baru saat tiba di sumber, tanpa memproses ulang file yang sudah ada. Dataset samples.sec.contracts dalam contoh ini bersifat statis, tetapi dengan sumber langsung, file baru akan terdeteksi setiap kali pipeline diperbarui. Untuk juga menyebarkan perubahan dan penghapusan sumber, masukkan feed perubahan dengan AUTO CDC. Lihat Terapkan pembaruan dan penghapusan dengan AUTO CDC.

Langkah 2. Perak: mengurai dan mengklasifikasikan dokumen

Teruskan setiap FILE ke fungsi ai_parse_document untuk mengonversi PDF mentah menjadi VARIANT terstruktur yang berisi elemen dokumen, metadata tata letak, dan teks. Karena ai_parse_document menerima kolom FILE , ia membaca dokumen langsung dari penyimpanan dan tidak pernah memuat byte ke memori cluster.

SQL

CREATE OR REFRESH MATERIALIZED VIEW parsed_contracts AS
  SELECT
    path,
    ai_parse_document(file) AS parsed
  FROM raw_contracts;

Python

@dp.materialized_view(name="parsed_contracts")
def parsed_contracts():
  return (
    spark.read.table("raw_contracts")
      .selectExpr("path", "ai_parse_document(file) AS parsed")
  )

Catatan

Mendefinisikan langkah parse sebagai tampilan materialisasi di atas raw_contracts tabel streaming akan meningkatkan perhitungan. Setiap pembaruan pipeline menjalankan ai_parse_document hanya pada file yang ditambahkan sejak pembaruan terakhir, bukan pada seluruh tabel. Karena ai_parse_document ini adalah langkah yang paling mahal, ini menghindari perbaikan dokumen yang sudah Anda proses. Penyegaran inkremental tampilan materialisasi membutuhkan komputasi serverless; Jalankan pipeline di serverless. Lihat Spark Declarative Pipelines di Lakeflow.

Selanjutnya, teruskan output yang telah diurai ke ai_classify fungsi untuk menetapkan setiap dokumen salah satu dari lima jenis perjanjian. Dokumen dengan kesalahan penguraian difilter sebelum klasifikasi. Contoh ini mengunci ai_classify ke versi 2.1, yang mengembalikan klasifikasi sebagai objek per label, jadi baca label dari kunci value.

SQL

CREATE OR REFRESH MATERIALIZED VIEW classified_contracts AS
  SELECT
    path,
    parsed,
    ai_classify(
      parsed,
      '["affiliate_agreement", "marketing_agreement", "consulting_agreement", "hosting_agreement", "escrow_agreement"]',
      map('version', '2.1')
    ):response[0].value::STRING AS contract_type
  FROM parsed_contracts
  WHERE is_variant_null(parsed:error_status);

Python

@dp.materialized_view(name="classified_contracts")
def classified_contracts():
  return (
    spark.read.table("parsed_contracts")
      .filter("is_variant_null(parsed:error_status)")
      .selectExpr(
        "path",
        "parsed",
        """ai_classify(
             parsed,
             '["affiliate_agreement", "marketing_agreement", "consulting_agreement", "hosting_agreement", "escrow_agreement"]',
             map('version', '2.1')
           ):response[0].value::STRING AS contract_type""")
  )

Tip

Untuk meningkatkan akurasi klasifikasi, tambahkan deskripsi label dan instructions opsi ke ai_classify. Silakan lihat fungsi ai_classify.

Langkah 3. Gold: ekstrak kolom per jenis perjanjian

Setiap jenis perjanjian memiliki set bidang relevan sendiri. Saring dokumen yang telah diklasifikasikan menjadi satu jenis, teruskan konten yang telah diurai ke fungsi ai_extract dengan skema bidang yang Anda inginkan, lalu ubah respons menjadi kolom bertipe. Contoh ini menetapkan ai_extract ke versi 2.1, di mana setiap kolom yang diekstrak berupa objek, jadi baca kunci value-nya.

Contoh berikut membangun tabel emas untuk perjanjian konsultasi:

SQL

CREATE OR REFRESH MATERIALIZED VIEW consulting_agreements AS
  WITH extracted AS (
    SELECT
      path,
      ai_extract(
        parsed,
        '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
        map('version', '2.1')
      ) AS fields
    FROM classified_contracts
    WHERE contract_type = 'consulting_agreement'
  )
  SELECT
    path,
    fields:response.company_name.value::STRING AS company_name,
    fields:response.consultant_name.value::STRING AS consultant_name,
    fields:response.compensation_amount.value::STRING AS compensation_amount,
    fields:response.effective_date.value::STRING AS effective_date
  FROM extracted;

Python

@dp.materialized_view(name="consulting_agreements")
def consulting_agreements():
  return (
    spark.read.table("classified_contracts")
      .filter("contract_type = 'consulting_agreement'")
      .selectExpr(
        "path",
        """ai_extract(
             parsed,
             '["company_name", "consultant_name", "compensation_amount", "effective_date"]',
             map('version', '2.1')
           ) AS fields""")
      .selectExpr(
        "path",
        "fields:response.company_name.value::STRING AS company_name",
        "fields:response.consultant_name.value::STRING AS consultant_name",
        "fields:response.compensation_amount.value::STRING AS compensation_amount",
        "fields:response.effective_date.value::STRING AS effective_date")
  )

Dengan perintah ini, Anda memiliki pipeline yang sepenuhnya inkremental: saat PDF kontrak baru tiba di volume, Auto Loader mengingesnya sebagai referensi terkelola FILE, ai_parse_document lalu ai_classify mengarahkan setiap dokumen, dan consulting_agreements tampilan materialisasi emas akan menampilkan field yang diekstrak.

Contoh buku catatan

Notebook berikut berisi seluruh pipeline dari tutorial ini. Notebook ini adalah kode sumber pipeline, bukan notebook yang dapat dijalankan. Impor notebook untuk bahasa Anda, lalu tentukan jalurnya di kolom Source code saat Anda mengonfigurasi pipeline. Lihat Mengonfigurasi alur.

SQL

Notebook SQL pipeline pemrosesan file

Dapatkan buku catatan

Python

Notebook Python untuk pipeline pemrosesan file

Dapatkan buku catatan

Jelajahi sendiri

Pipeline mengklasifikasikan dokumen menjadi lima jenis perjanjian, tetapi hanya mengekstrak kolom untuk consulting_agreement. Untuk memperluasnya, ulangi langkah emas untuk setiap tipe yang tersisa, ubah contract_type filter dan ai_extract skema agar sesuai dengan bidang yang relevan dengan tipe tersebut. Contohnya:

affiliate_agreement: , party_1_name, party_2_name, commission_rate, payment_frequency
marketing_agreement: , party_1_name, party_2_name, effective_date, territory
hosting_agreement: , provider_name, customer_name, effective_date, term_length
escrow_agreement: , owner_name, licensee_name, escrow_agent_name, software_name

Sumber daya tambahan

FILE jenis
Memasukkan file sebagai tipe FILE
Fungsi FILE mulai cepat
Pelajari selengkapnya tentang Auto Loader. Lihat Apa itu Auto Loader?.

Saran dan Komentar

Apakah halaman ini membantu?

Terakhir diperbarui pada 2026-10-09

Tutorial: Membangun pipeline pemrosesan file dengan tipe FILE

Persyaratan

Buat pipeline pemrosesan file

Langkah 1. Perunggu: mengimpor PDF mentah sebagai referensi FILE yang dikelola

SQL

Python

Langkah 2. Perak: mengurai dan mengklasifikasikan dokumen

SQL

Python

SQL

Python

Langkah 3. Gold: ekstrak kolom per jenis perjanjian

SQL

Python

Contoh buku catatan

SQL

Notebook SQL pipeline pemrosesan file

Python

Notebook Python untuk pipeline pemrosesan file

Jelajahi sendiri

Sumber daya tambahan

Saran dan Komentar

Sumber Daya Tambahan: