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

Important

Bu özellik Beta sürümündedir. Bunu kullanmak için, bir çalışma alanı yöneticisi Önizlemeler sayfasından Dosya tipi seçeneğini açmalıdır. Bkz. Azure Databricks önizlemelerini yönetme.

Bu eğitimde, yapılandırılmamış belgeleri uçtan uca işleyen Lakeflow boru hattıyla bir madalyon sistemi inşa ediyorsunuz. Bu örnek, Unity Catalog biriminde PDF olarak depolanan, SEC’e sunulmuş yasal sözleşmelerden oluşan samples.sec.contracts örnek veri kümesini kullanır.

İş hattı, PDF'leri Auto Loader ile yönetilen FILE referanslar olarak içeri 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 iş hattı ortaya çıkar: bronz (ham yönetilen FILE referanslar), gümüş (ayrıştırılmış ve sınıflandırılmış 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.
  • Çalışma alanınız için FILE türünü etkinleştirin. Çalışma alanı yöneticileri bunu Önizlemeler sayfasından etkinleştirebilir. Bkz. Azure Databricks önizlemelerini yönetme.
  • Bir şemada tablo oluşturma ve bir pipeline oluşturma iznlerine sahip olmak.
  • Yazabileceğiniz bir Unity Kataloğu birimi olsun. Bu birimi bronz tablonun FileSpaceolarak bildirirsiniz ve Unity Catalog alınan dosyaları bunun içine yönetilen depolama alanı olarak kopyalar.
  • Önizleme kanalını kullanın.

Veri samples.sec.contracts seti varsayılan olarak tüm çalışma alanlarında mevcuttur. Bu eğitim, içeri aktarılan PDF'leri referans olarak FILE MANAGED saklar: Unity Catalog, her dosyayı tablonun FileSpace olarak belirttiğiniz birime kopyalar ve onu tabloyla birlikte yönetir; böylece satırlar silindiğinde referans verilen dosyalar çöp toplamaya uygun hale gelir ve tablo ve dosyaları senkronize kalır. İş hattını kendi PDF'lerinize uyarlamak için kaynak yolunu dosyalarınızı içeren bir birime yönlendirin. Diğer alım seçenekleri için Dosyaları FILE türü olarak içe aktarma bölümüne bakın.

Dosya işleme boru hattı oluşturun

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

Adım 1. Bronze: ham PDF'leri yönetilen DOSYA referansları olarak içe aktarı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 MANAGED olarak bildirilmesi, her dosyayı tablonun FileSpace öğesine, yani databricks.filespace-preview tablo özelliğiyle ayarladığınız hacme kopyalar; böylece Unity Catalog dosyaları tabloyla birlikte yönetir.

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/")
  )
  • Büyük dosyalar için çalışır: büyük bir PDF tablonun FileSpace içinde bulunurken, tablo satırı yalnızca hafif bir FILE referans (uri, size, content_type, checksum) saklar. Bunu, baytları satır içinde tutan BINARY türüyle karşılaştırın.
  • Yönetilen dosya yaşam döngüsü: Unity Catalog, alınan her dosyayı tablonun FileSpace içine kopyalar ve tablo ile birlikte yönetir: satırlar silindiğinde, başvurulan dosyalar çöp toplamaya uygun hale gelir, böylece tablo ve dosyaları eşzamanlı kalır. Ayrıntılar için bkz. YÖNETİLEN DOSYA ve DOSYA HARİCİ.
  • 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şikliklerini ve silme işlemlerini de iletmek için, değişiklik akışını AUTO CDC ile alı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. FILE, ai_parse_document sütununu kabul ettiğinden belgeyi doğrudan depolama alanından okur ve baytları hiçbir zaman 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ı raw_contracts akış tablosu üzerinde somutlaştırılmış bir görünüm olarak tanımlamak, hesaplamayı artımlı hâle getirir. İşlem hattındaki her güncellemede ai_parse_document, tablonun tamamında değil, yalnızca son güncellemeden bu yana eklenen dosyalara uygulanır. ai_parse_document en maliyetli adım olduğundan, bu işlem daha önce işlediğiniz belgeleri yeniden ayrıştırmanızı önler. Maddeleştirilmiş görünümlerin kademeli yenilenmesi, sunucusuz hesaplama gerektirir; Pipeline'ı sunucusuz çalıştırın. Lakeflow üzerindeki Spark Declarative Pipelines bölümüne bakın.

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, ai_classify öğesini sınıflandırmayı etiket başına bir nesne olarak döndüren 2.1 sürümüne sabitler; bu nedenle etiketi value anahtarından okuyun.

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 ai_classify öğesine etiket açıklamaları ve instructions seçeneği ekleyin. 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. Sınıflandırılmış belgeleri tek bir türle sınırlandırın, ayrıştırılmış içeriği istediğiniz alanların şemasıyla birlikte ai_extract işlevine iletin, ardından yanıtı türlendirilmiş sütunlara düzleştirin. Bu örnek, value öğesini 2.1 sürümüne sabitler; bu sürümde çıkarılan her alan bir nesnedir, bu yüzden ai_extract anahtarını okuyun.

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 artımlı bir işlem hattınız olur: yeni sözleşme PDF'leri birime geldiğinde, Auto Loader bunları yönetilen FILE başvurular olarak alır, ai_parse_document ve ai_classify her belgeyi yönlendirir ve consulting_agreements altın somutlaştırılmış görünüm çıkarılan alanları ortaya çıkarır.

Örnek not defterleri

Aşağıdaki notebook'lar, bu eğitimdeki tüm iş hattını içermektedir. Bu defterler pipeline kaynak kodudur, çalıştırılabilir not defterleri değildir. Dilinize ait notebook'u içe aktarın, ardından işlem hattını yapılandırırken Kaynak kodu alanında yolunu belirtin. Bkz . İşlem hatlarını yapılandırma.

SQL

Dosya işleme pipeline SQL notebook

Dizüstü bilgisayar al

Python

Python dosya işleme iş hattı not defteri

Dizüstü bilgisayar al

Kendi başınıza keşfedin

Pipeline, belgeleri beş anlaşma türünden birine sınıflandırır, ancak yalnızca consulting_agreement için alanları ayıklar. Bunu genişletmek için, kalan her tür için altın adımı tekrarlayın; contract_type filtresini ve ai_extract şemasını, o türle ilgili alanlara uyacak şekilde 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

  • Son güncelleştirme: