Instalace privátního balíčku jako požadavku v úloze Apache Airflow

Poznámka:

Úloha Apache Airflow využívá Apache Airflow.

Balíček Pythonu je způsob, jak uspořádat související moduly Pythonu do jedné hierarchie adresářů. Balíček je obvykle reprezentován jako adresář, který obsahuje speciální soubor s názvem init.py. V adresáři balíčku můžete mít několik souborů modulu Pythonu (.py souborů), které definují funkce, třídy a proměnné. V kontextu úlohy Apache Airflow můžete vyvíjet privátní balíčky pro přidání vlastních operátorů Apache Airflow, háčků, senzorů, modulů plug-in atd.

V tomto kurzu vytvoříte jednoduchý vlastní operátor jako balíček Pythonu, přidáte ho jako požadavek v prostředí úloh Apache Airflow a naimportujete privátní balíček jako modul v souboru DAG.

Vývoj vlastního operátoru a testování pomocí Apache Airflow Dag

1. Vytvořte soubor sample_operator.py a převeďte ho na privátní balíček. Projděte si průvodce: Vytvoření balíčku v Pythonu

   from airflow.models.baseoperator import BaseOperator
   
   
   class SampleOperator(BaseOperator):
       def __init__(self, name: str, **kwargs) -> None:
           super().__init__(**kwargs)
           self.name = name
   
       def execute(self, context):
           message = f"Hello {self.name}"
           return message
   
   

2. Vytvořte soubor sample_dag.py DAG Apache Airflow, který otestuje operátor definovaný v kroku 1.

   from datetime import datetime
   from airflow import DAG
   
    # Import from private package
   from airflow_operator.sample_operator import SampleOperator
   
   
   with DAG(
   "test-custom-package",
   tags=["example"]
   description="A simple tutorial DAG",
   schedule_interval=None,
   start_date=datetime(2021, 1, 1),
   ) as dag:
       task = SampleOperator(task_id="sample-task", name="foo_bar")
   
       task
   

3. Vytvořte úložiště GitHub obsahující složku a Dags soubor vašeho privátního sample_dag.py balíčku. Mezi běžné formáty souborů patří zip, .whlnebo tar.gz. Podle potřeby soubor umístěte do složky Dags nebo Plugins. Synchronizace úložiště Git s úlohou Apache Airflow nebo můžete použít předkonfigurované úložištěInstall-Private-Package.

Přidání balíčku jako požadavku

Přidejte balíček jako požadavek v části Airflow requirements. Použití formátu /opt/airflow/git/<repoName>.git/<pathToPrivatePackage>

Pokud se například váš privátní balíček nachází v /dags/test/private.whl úložišti GitHubu, přidejte požadavek /opt/airflow/git/<repoName>.git/dags/test/private.whl do prostředí Airflow.

Související obsah

Rychlý start: Vytvoření úlohy Apache Airflow