Samouczek: Zbuduj potok przetwarzania plików z typem PLIKU

W tym samouczku budujesz system medallion z potokiem Lakeflow, który przetwarza nieustrukturyzowane dokumenty od początku do końca. Ten przykład wykorzystuje przykładowy zbiór danych samples.sec.contracts — kolekcję umów prawnych zgłoszonych do SEC, przechowywanych jako pliki PDF w woluminie Unity Catalog.

Potok pobiera pliki PDF jako zarządzane FILE referencje za pomocą Auto Loadera, analizuje każdy dokument za pomocą funkcji AI, klasyfikuje go do typu umowy i wyodrębnia pola strukturalne dla każdego typu.

Aby uzyskać informacje o typie, zobacz FILE typ.

Ten samouczek obejmuje następujące kroki:

• Stopniowo importuj pliki PDF kontraktów z woluminu jako zarządzane FILE referencje za pomocą Auto Loadera.
• Parsuj każdy dokument z funkcjąai_parse_document i klasyfikuj go według funkcjiai_classify.
• Wyodrębnij pola strukturalne dla każdego typu umowy z ai_extract funkcją.

Efektem jest potok danych w stylu medalionowym: brąz (surowe FILE zarządzane odniesienia), srebro (przeanalizowane i sklasyfikowane dokumenty) oraz złoto (wydobyte pola według typu umowy). Aby uzyskać więcej informacji, zobacz Co to jest architektura medallion lakehouse? Warstwa brązu to tabela strumieniowa , która stopniowo pobiera pliki, a warstwy srebrnej i złotej to zmaterializowane widoki , które przetwarzają się dopiero wtedy, gdy ich dane się zmieniają.

Requirements

Aby ukończyć ten samouczek, musisz spełnić następujące wymagania:

• Zaloguj się do przestrzeni roboczej Azure Databricks z włączonym Unity Catalog.
• Włącz ten typ FILE dla swojego workspace. Administratorzy przestrzeni roboczej mogą to włączyć ze strony Zapowiedzi . Zobacz Zarządzanie wersjami zapoznawczami usługi Azure Databricks.
• Posiadanie uprawnień do tworzenia tabel w schemacie oraz do tworzenia pipeline.
• Miej wolumin Unity Catalog, do którego możesz pisać. Deklarujesz ten wolumin jako tabelę bronzeFileSpace, a Unity Catalog kopiuje do niej pozyskane pliki jako zarządzaną pamięć.
• Użyj kanału Podgląd.

Zbiór samples.sec.contracts danych jest domyślnie dostępny we wszystkich przestrzeniach roboczych. Ten samouczek przechowuje pozyskane pliki PDF jako odwołania FILE MANAGED: Unity Catalog kopiuje każdy plik do woluminu zadeklarowanego jako FileSpace tabeli i zarządza nim razem z tabelą, więc usunięcie wierszy sprawia, że pliki, do których prowadzą odwołania, kwalifikują się do garbage collection, a tabela i jej pliki pozostają zsynchronizowane. Aby dostosować pipeline do własnych plików PDF, skieruj ścieżkę źródłową na wolumin zawierający twoje pliki. Aby uzyskać informacje o innych opcjach pozyskiwania, zobacz Pozyskiwanie plików jako typ FILE.

Utwórz potok przetwarzania plików

Pipeline przetwarza dokumenty w trzech etapach.

Krok nr 1. Brąz: wczytywanie surowych plików PDF jako zarządzanych odwołań FILE

Użyj Auto Loader, aby przyrostowo odczytywać pliki PDF umów z woluminu. Odczytywanie plików za pomocą format => 'file' powoduje przechwycenie odwołania i metadanych każdego pliku bez wczytywania jego zawartości w postaci bajtów. Zadeklarowanie kolumny jako FILE MANAGED kopiuje każdy plik do elementu FileSpace tabeli, ustawianego za pomocą właściwości tabeli databricks.filespace-preview, więc Unity Catalog zarządza plikami razem z tabelą.

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

• Działa dla dużych plików: duży plik PDF znajduje się w tabeli FileSpace, podczas gdy wiersz tabeli przechowuje tylko lekkie FILE odwołanie (uri, size, , content_type, checksum). Porównaj to z typem BINARY , który wplata bajty w wierszu.
• Zarządzany cykl życia plików: Unity Catalog kopiuje każdy pozyskany plik do lokalizacji tabeli FileSpace i zarządza nim za pomocą tabeli: usunięcie wierszy sprawia, że odwołane pliki kwalifikują się do garbage collection, dzięki czemu tabela i jej pliki pozostają zsynchronizowane. Szczegóły można znaleźć w FILE MANAGED i FILE EXTERNAL.
• Przetwarzanie przyrostowe: tabela strumieniowa stopniowo pobiera nowe pliki, gdy dotrą do źródła, bez ponownego przetwarzania istniejących. Zbiór samples.sec.contracts danych w tym przykładzie jest statyczny, ale przy live source nowe pliki są pobierane przy każdej aktualizacji pipeline. Aby propagować również zmiany i usunięcia w źródle, należy pozyskiwać strumień zmian za pomocą AUTO CDC. Zobacz Zastosuj aktualizacje i usunięcia za pomocą AUTO CDC.

Krok nr 2. Srebro: analizować i klasyfikować dokumenty

Każdy FILE przekaż do funkcji ai_parse_document, aby przekonwertować nieprzetworzony plik PDF na ustrukturyzowany VARIANT zawierający elementy dokumentu, metadane układu i tekst. Ponieważ ai_parse_document akceptuje kolumnę FILE , odczytuje dokument bezpośrednio z pamięci i nigdy nie ładuje bajtów do pamięci klastrowej.

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

Następnie przekaż wynik parsowania do ai_classifyfunkcji, aby przypisać każdemu dokumentowi jeden z pięciu typów umów. Dokumenty z błędami analizy są filtrowane przed klasyfikacją. Ten przykład przypisuje element ai_classify do wersji 2.1, która zwraca klasyfikację jako obiekt osobno dla każdej etykiety, dlatego etykietę należy odczytać z klucza 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""")
  )

Krok nr 3. Złoto: pola ekstrakcyjne według typu umowy

Każdy typ umowy ma własny zestaw odpowiednich pól. Przefiltruj sklasyfikowane dokumenty do jednego typu, przekaż przeanalizowaną zawartość do ai_extractfunkcji ze schematem wybranych pól, a następnie spłaszcz odpowiedź do kolumn typowanych. Ten przykład przypina ai_extract do wersji 2.1, w której każde wyodrębnione pole jest obiektem, więc odczytaj jego value klucz.

Poniższy przykład tworzy złotą tabelę dla umów konsultingowych.

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

Dzięki tym instrukcjom masz w pełni inkrementalny pipeline: gdy nowe PDF-y kontraktów pojawiają się w woluminie, Auto Loader wprowadza je jako zarządzane FILE referencje ai_parse_document, a ai_classify kieruje każdy dokument, a consulting_agreements zmaterializowany widok gold wyświetla wyodrębnione pola.

Przykładowe notatniki

Poniższe notebooki zawierają pełny pipeline z tego samouczka. Te notebooki to kod źródłowy pipeline, a nie notatniki do uruchomienia. Zaimportuj notebook dla swojego języka, a następnie określ jego ścieżkę w polu Kod źródłowy podczas konfiguracji pipeline. Zobacz Konfigurowanie potoków.

SQL

Notes SQL potoku przetwarzania plików

Pobierz laptopa

Python

Notatnik Python do przetwarzania plików

Pobierz laptopa

Eksploruj na własną rękę

Proces klasyfikuje dokumenty do pięciu typów umów, ale wyodrębnia pola tylko dla consulting_agreement. Aby to rozszerzyć, powtórz krok gold dla każdego z pozostałych typów, zmieniając filtr contract_type i schemat ai_extract tak, aby odpowiadały polom istotnym dla danego typu. Przykład:

• affiliate_agreement: party_1_name, , party_2_name, , commission_ratepayment_frequency
• marketing_agreement: party_1_name, , party_2_name, , effective_dateterritory
• hosting_agreement: provider_name, , customer_name, , effective_dateterm_length
• escrow_agreement: owner_name, , licensee_name, , escrow_agent_namesoftware_name

Dodatkowe zasoby

• FILE typ
• Importuj pliki jako typ FILE
• Krótkie wprowadzenie do funkcji FILE
• Dowiedz się więcej o automatycznym ładowaniu. Zobacz Co to jest moduł automatycznego ładowania?.