Provozování pipelinů ve workflow

Kanál můžete spustit jako součást pracovního postupu zpracování dat pomocí úloh Lakeflow, Apache Airflow nebo Azure Data Factory.

Pipeline automaticky řeší závislosti mezi svými datovými sadami, takže sama zvládá jednoduchou orchestraci v rámci pipeline. Pro orchestraci, pro kterou není kanál sestavený, jako je podmíněné spouštění, větvení výsledků úkolů, opakování nebo koordinace kanálu s jinými typy práce, použijte místo vytváření logiky do kanálu vyhrazený orchestrátor pracovního postupu.

Připravte svou pipeline pro orchestraci

Orchestrace funguje nejlépe, když každý kanál pokrývá samostatnou jednotku práce, kterou chcete naplánovat, ověřit nebo spustit samostatně. Navrhněte kanály kolem těchto hranic tak, aby je pracovní postup mohl koordinovat jako samostatné úkoly, včetně vhodného toku řízení mezi nadřazenými a podřízenými úkoly.

Pokud už máte velkou pipeline, která sdružuje úlohy, jež chcete spravovat samostatně, rozdělte ji na menší pipeline přesunutím tabulek do nové pipeline. Viz Přesun tabulek mezi kanály.

Úlohy Lakeflow

V úlohách Lakeflow můžete orchestrovat několik úloh a implementovat pracovní postup zpracování dat. Pokud chcete do úlohy zahrnout kanál, při vytváření úlohy použijte úlohu Kanálu . Viz úloha kanálu pro úlohy.

Apache Airflow

Apache Airflow je opensourcové řešení pro správu a plánování pracovních postupů dat. Airflow představuje pracovní postupy jako řízené acyklické grafy (DAG) operací. Pracovní postup definujete v souboru Pythonu a Airflow spravuje plánování a provádění. Informace o instalaci a používání Airflow s Azure Databricks najdete v tématu Orchestrate Lakeflow Jobs with Apache Airflow.

Pokud chcete spustit pipeline jako součást pracovního postupu Airflow, použijte DatabricksSubmitRunOperator.

Požadavky

Pro použití podpory Airflow pro kanály Lakeflow jsou potřeba následující:

Example

Následující příklad vytvoří Airflow DAG, který aktivuje aktualizaci pipeline s identifikátorem 8279d543-063c-4d63-9926-dae38e35ce8b:

from airflow import DAG
from airflow.providers.databricks.operators.databricks import DatabricksSubmitRunOperator
from airflow.utils.dates import days_ago

default_args = {
  'owner': 'airflow'
}

with DAG('ldp',
         start_date=days_ago(2),
         schedule_interval="@once",
         default_args=default_args
         ) as dag:

  opr_run_now=DatabricksSubmitRunOperator(
    task_id='run_now',
    databricks_conn_id='CONNECTION_ID',
    pipeline_task={"pipeline_id": "8279d543-063c-4d63-9926-dae38e35ce8b"}
  )

Nahraďte CONNECTION_ID identifikátorem připojení Airflow k vašemu pracovnímu prostoru.

Uložte tento příklad do airflow/dags adresáře a pomocí uživatelského rozhraní Airflow zobrazte a aktivujte DAG. Pomocí uživatelského rozhraní pipeline můžete zobrazit podrobnosti o aktualizaci tohoto procesu.

Azure Data Factory

Poznámka:

Kanály nástroje Lakeflow i Azure Data Factory nabízejí možnosti nastavení počtu opakování, když dojde k selhání. Pokud jsou hodnoty opakování nakonfigurované ve vašem kanálu a v aktivitě služby Azure Data Factory, která kanál volá, je počet opakování vynásobenou hodnotou opakování kanálu.

Pokud například aktualizace kanálu selže, kanál ve výchozím nastavení aktualizaci opakuje až pětkrát. Pokud je opakování služby Azure Data Factory nastavené na tři a váš kanál používá výchozí nastavení pěti opakování, může se váš neúspěšný kanál opakovat až patnáctkrát. Aby nedocházelo k nadměrným pokusům o opakování při selhání aktualizací kanálu, doporučuje Databricks omezit počet opakování při konfiguraci kanálu nebo aktivity služby Azure Data Factory, která kanál volá.

Pokud chcete změnit nastavení opakování pro vaši pipeline, použijte pipelines.numUpdateRetryAttempts při konfiguraci pipeline.

Azure Data Factory je cloudová služba ETL, která umožňuje orchestraci pracovních postupů integrace a transformace dat. Azure Data Factory přímo podporuje spouštění úloh Azure Databricks v pracovním postupu, včetně poznámkových bloků, úloh JAR a skriptů Pythonu. Do pracovního postupu můžete datový tok zahrnout také použitím rozhraní REST API z webové aktivity služby Azure Data Factory. Pokud například chcete spustit aktualizaci pipeline ze služby Azure Data Factory:

  1. Vytvořte datovou továrnu nebo otevřete existující datovou továrnu.

  2. Po dokončení vytváření otevřete stránku datové továrny a klikněte na dlaždici Otevřít Azure Data Factory Studio . Zobrazí se uživatelské rozhraní služby Azure Data Factory.

  3. Nový datový kanál v Azure Data Factory vytvoříte tak, že v uživatelském rozhraní Azure Data Factory Studio vyberete datový kanál z rozevírací nabídky Nový.

  4. Na panelu nástrojů Aktivity rozbalte Obecné a přetáhněte Webovou aktivitu na plátno potrubí. Klikněte na kartu Nastavení a zadejte následující hodnoty:

    Poznámka:

    Jako bezpečnostní opatření při ověřování pomocí automatizovaných nástrojů, systémů, skriptů a aplikací doporučuje Databricks používat osobní přístupové tokeny, které patří služebním subjektům, místo uživatelů pracovního prostoru. Pokud chcete vytvořit tokeny pro instanční objekty, přečtěte si téma Správa tokenů instančního objektu.

    • Adresa URL: https://<databricks-instance>/api/2.0/pipelines/<pipeline-id>/updates.

      Nahradit <get-workspace-instance>.

      Nahraďte <pipeline-id> identifikátorem kanálu.

    • Metoda: V rozevírací nabídce vyberte POST .

    • Záhlaví: Klikněte na + Nový. Do textového pole Název zadejte Authorization. Do textového pole Hodnota zadejte Bearer <personal-access-token>.

      Nahraďte <personal-access-token>osobním přístupovým tokenem Azure Databricks.

    • Text: Pokud chcete předat další parametry požadavku, zadejte dokument JSON obsahující parametry. Chcete-li například spustit aktualizaci a znovu zpracovat všechna data pro datový tok: {"full_refresh": "true"}. Pokud neexistují žádné další parametry požadavku, zadejte prázdné složené závorky ({}).

Chcete-li otestovat webovou aktivitu, klepněte na tlačítko Ladit na panelu nástrojů v uživatelském rozhraní služby Data Factory. Výstup a stav spuštění, včetně chyb, se zobrazí na kartě Výstup v Azure Data Factory pipeline. K zobrazení podrobností o aktualizaci kanálu použijte uživatelské rozhraní kanálů.

Návod

Běžným požadavkem pracovního postupu je zahájení úkolu po dokončení předchozího úkolu. Vzhledem k tomu, že požadavek kanálu updates je asynchronní a vrací se po spuštění aktualizace, ale před dokončením, musí úlohy v kanálu Azure Data Factory se závislostí na aktualizaci kanálu čekat, než se aktualizace dokončí. Možnost, jak počkat na dokončení aktualizace, spočívá v přidání aktivity Until po webové aktivitě, která spouští aktualizaci kanálu. V aktivitě Until:

  1. Přidejte aktivitu čekání , která po dokončení aktualizace čeká na nakonfigurovaný počet sekund.
  2. Přidejte webovou aktivitu za aktivitou Čekání, která používá požadavek na podrobnosti aktualizace kanálu k získání stavu aktualizace. Pole state v odpovědi vrátí aktuální stav aktualizace, včetně toho, jestli byla dokončena.
  3. Pomocí hodnoty state pole nastavte ukončovací podmínku aktivity Until. Aktivitu Nastavit proměnnou můžete použít také k přidání proměnné kanálu na základě state hodnoty a k použití této proměnné pro podmínku ukončení.
  • Last updated on