Eğitim: FILE tipiyle bir dosya işleme boru hattı oluşturun

Important

Bu özellik Beta sürümündedir. Çalışma alanı yöneticileri Bu özelliğe erişimi Önizlemeler sayfasından denetleyebilir. Bkz. Azure Databricks önizlemelerini yönetme.

Lakeflow boru hattıyla yapılandırılmamış belgeleri uçtan uca işleyen bir madalyon boru hattı nasıl inşa edileceğini öğrenin. Bu örnek, samples.sec.contracts SEC tarafından doldurulmuş yasal anlaşmaların PDF olarak Unity Kataloğu ciltinde saklanan örnek veri setini kullanır.

Boru hattı, PDF'leri Otomatik Yükleyici ile harici FILE referans olarak alır, her belgeyi yapay zeka fonksiyonlarıyla ayrıştırır, bir anlaşma türüne sınıflandırır ve her tür için yapılandırılmış alanlar çıkarır.

Tip referansı için bkz.FILE

Bu kılavuzda şunları yapacaksınız:

Sonuç olarak madalyon tarzı bir boru hattı ortaya çıkar: bronz (ham dış FILE referanslar), gümüş (ayrıştırılmış ve gizli belgeler) ve altın (anlaşma türüne göre çıkarılmış alanlar). Daha fazla bilgi için bkz. Madalyon göl evi mimarisi nedir? Bronz katman, dosyaları kademeli olarak alan bir akış tablosudur ve gümüş ve altın katmanlar yalnızca girdileri değiştiğinde yeniden hesaplanan maddi görünümlerdir .

Requirements

Bu öğreticiyi tamamlamak için aşağıdaki gereksinimleri karşılamanız gerekir:

  • Unity Catalog etkin iken bir Azure Databricks çalışma alanına giriş yapın.
  • Bir şemada tablo oluşturma ve bir pipeline oluşturma iznlerine sahip olmak.
  • Önizleme kanalını kullanın.

Veri samples.sec.contracts seti varsayılan olarak tüm çalışma alanlarında mevcuttur, bu yüzden ek kurulum gerekmez. Dosyalar zaten bir Unity Kataloğu ciltinde bulunduğundan, bu eğitim onları FILE EXTERNAL referans olarak saklıyor ve içeriklerini kopyalamıyor. Boru hattını kendi PDF'lerinize uyarlamak için, kaynak yolunu dosyalarınızı içeren bir hacme yönlendirin. Diğer alım seçenekleri için FILE tipi olarak Ingest files sayfasına bakınız.

Dosya işleme boru hattı oluşturun

Boru hattı belgeleri üç aşamada işliyor.

Adım 1. Bronze: ham PDF'leri harici FILE referansları olarak alın

Ciltten sözleşme PDF'lerini kademeli olarak okumak için Auto Loader kullanın. Dosyaları okumak format => 'file' , her dosyanın baytlarını oluşturmadan bir referans ve meta veri yakalar. Sütunun FILE EXTERNAL referans olarak ilan edilmesi, her dosyanın içeriğini kopyalamadan yerinde olmasını sağlar.

SQL

CREATE OR REFRESH STREAMING TABLE raw_contracts (
  path STRING,
  size BIGINT,
  modification_time TIMESTAMP,
  file FILE EXTERNAL
)
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 EXTERNAL"
)
def raw_contracts():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "file")
      .load("/Volumes/samples/sec/contracts/")
  )
  • Büyük dosyalar için çalışır: büyük bir PDF hacimde kalırken, tablo satırı yalnızca hafif FILE bir referans (uri, size, content_type, checksum) depolar. Bunu, sıradaki baytların iç çizimini oluşturan type BINARY ile karşılaştırın.
  • Artan işleme: Akış tablosu, mevcut dosyaları yeniden işlemeden kaynağa ulaştıkça yeni dosyaları kademeli olarak alır. Bu örnekteki samples.sec.contracts veri seti statiktir, ancak canlı bir kaynakla her boru hattı güncellemesinde yeni dosyalar alınır. Kaynak değişiklikleri ve silmeleri de yaymak için, değişim beslemesini ile AUTO CDCalın. Bkz. Güncellemeleri ve silmeleri AUTO CDC ile uygula.

Adım 2. Gümüş: belgeleri ayrıştırıp sınıflandır

Her FILEai_parse_document birini fonksiyona aktararak ham PDF'yi belge öğeleri, düzen meta verileri ve metin içeren yapılandırılmış VARIANT bir hale dönüştürün. Bir FILE sütun kabul ettiği ai_parse_document için, belgeyi doğrudan depolamadan okur ve baytları küme belleğine yüklemez.

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

Uyarı

Ayrıştırma adımını, akış raw_contracts tablosu üzerinde maddi bir görünüm olarak tanımlamak, hesaplamayı artırarak artırır. Her pipeline güncellemesi ai_parse_document sadece son güncellemeden sonra eklenen dosyalarda çalışır, tüm tabloda değil. En pahalı adım olduğu ai_parse_document için, daha önce işlediğiniz belgeleri yeniden incelemeden kaçınır. Maddeleştirilmiş görünümlerin kademeli yenilenmesi, sunucusuz hesaplama gerektirir; Pipeline'ı sunucusuz çalıştırın. Bkz. Spark Bildirimsel İşlem Hatları.

Sonra, her belgeye beş anlaşma türünden birini atamak için ayrıştırılmış çıktıyı ai_classify fonksiyona iletin. Ayrıştırma hataları olan belgeler sınıflandırmadan önce filtrelenir. Bu örnek, sınıflandırmayı etiket başına nesne olarak döndüren 2.1 sürümüne sabitler ai_classify , yani etiketi anahtardan value okun.

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

Sınıflandırma doğruluğunu artırmak için, etiket tanımları ve instructions bir seçenek ekleyin ai_classify. bkz. ai_classify işlevi.

Adım 3. Altın: anlaşma türüne göre ekstrakt alanları

Her anlaşma türünün kendine özgü ilgili alanlar seti vardır. Gizli belgeleri tek bir türe filtreleyin, ayrıştırılmış içeriği ai_extract istediğiniz alanların şemasına göre işleyecek şekilde geçirin, ardından cevabı yazılı sütunlara düzleştirin. Bu örnek, her çıkarılan alan bir nesne value olduğu ve anahtarını okuyacağınız 2.1 sürümüne sabitlerai_extract.

Aşağıdaki örnek, danışmanlık anlaşmaları için altın tabloyu oluşturur:

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

Bu ifadelerle tamamen kademeli bir boru hattı elde edilir: yeni sözleşme PDF'leri cilme geldiğinde, Auto Loader bunları harici FILE referans ai_parse_document olarak alır ve ai_classify her belgeyi yönlendirir, consulting_agreements altın maddeleştirilmiş görünüm çıkarılmış alanları ortaya çıkarır.

Kendi başınıza keşfedin

Pipeline belgeleri beş anlaşma türüne sınıflandırır, ancak sadece consulting_agreementiçin alanlar çıkarır. Bunu uzatmak için, kalan her tip için altın adımı tekrarlayın, filtreyi contract_type ve şemayı ai_extract o tipe ait alanlarla eşleştirerek değiştirin. Örneğin:

  • 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

Ek kaynaklar

  • Last updated on