Jak automatizovat kód pomocí Airflow DAG?

Automatizace kódu pomocí DAG je výkonným nástrojem pro efektivní správu úloh v rámci datových pracovních toků. V tomto případě se zaměříme na vytvoření DAG v Apache Airflow, který umožní automatické spuštění kódu každý den. Tento proces zahrnuje několik důležitých kroků, které je třeba splnit, aby byl váš kód správně integrován do Airflow prostředí. V rámci tohoto procesu se seznámíme s některými nástroji, které jsme dosud nezmiňovali, jako jsou operátory a další techniky pro vytváření DAG.

python
Copy code
import json
import pathlib
import airflow
import requests
import requests.exceptions as request_exceptions
from datetime import date
from airflow import DAG
from airflow.operators.bash import BashOperator
from airflow.operators.python import PythonOperator
from airflow.decorators import task
from datetime import datetime, timedelta
dag_owner = 'Kendrick'

Následně si definujeme Python funkci pro získání nejnovějšího obrázku z NASA Planetary API. Funkce volá API s naším API klíčem a ukládá obrázek na disk, přičemž název souboru obsahuje aktuální datum.

python
Copy code
def _get_pictures():

    pathlib.Path("/tmp/images").mkdir(parents=True, exist_ok=True)
    api_key = 'xxxxxxxxxxxxxxxxxxxxxxxxxxxxx'
    url = f'https://api.nasa.gov/planetary/apod?api_key={api_key}'
    response = requests.get(url).json()
    today_image = response['hdurl']
    with open(f'todays_image_{date.today()}.png', 'wb') as f:
        f.write(requests.get(today_image).content)

Dalším krokem je nastavení výchozích argumentů pro DAG. Tato nastavení určují, jaké parametry bude Airflow používat při volání každého úkolu v DAG. Definujeme parametry jako majitele DAG, zda úkoly závisí na minulosti, počet pokusů o opakování a zpoždění mezi pokusy.

python
Copy code
default_args = {

    'owner': dag_owner,
    'depends_on_past': False,
    'retries': 2,
    'retry_delay': timedelta(minutes=5)
}

Dále inicializujeme DAG objekt a definujeme jeho parametry. Použití klíčového slova with nám umožňuje definovat kontext DAG a zamezit opětovnému inicializování objektu při každém analýze DAG kódu. Takto definujeme DAG, který se spustí každý den:

python
Copy code
with DAG(
    dag_id='download_ASOD_image',
    default_args=default_args,
    description='download and notify',
    start_date=airflow.utils.dates.days_ago(0),
    schedule_interval='@daily',
    catchup=False,
    tags=['None']
):

python
Copy code
get_pictures = PythonOperator(
    task_id="get_pictures",
    python_callable=_get_pictures
)
notify = BashOperator(
    task_id="notify",
    bash_command='echo "Image for today has been added!"'
)

Nakonec definujeme pořadí vykonávání úkolů v DAG. Určíme, že úkol get_pictures se má vykonat jako první, a úkol notify se vykoná po něm.

python
Copy code
get_pictures >> notify

Když se podíváme na celý DAG, může to vypadat na první pohled složitěji než kód v Jupyter Notebooku, který jsme používali dříve. Většina kódu však bude opakujícími se částmi, které pomáhají Apache Airflow pochopit, jak má workflow fungovat. Jako odborníci na data a vývojáři si musíme být vědomi toho, že existuje spousta různých možností pro nastavení a správu úloh v Airflow. Airflow umožňuje nejen rozdělit úkoly na menší části, ale také je paralelně vykonávat, což značně zvyšuje efektivitu.

Instanciování DAG objektu je základem každého pracovního toku, a je tedy klíčové správně nastavit parametry DAGu tak, aby odpovídaly požadavkům konkrétního úkolu. Například výchozí argumenty DAG jako dag_owner, depends_on_past, retries, a retry_delay mohou mít velký vliv na to, jak Airflow provádí úkoly v případě chyb nebo opakovaných pokusů.

Airflow také umožňuje definovat interval, jak často se má DAG spouštět. Tento interval může být nastaven na různé hodnoty, jako je @daily, @weekly nebo i na základě cron výrazu. Pro tento konkrétní případ používáme denní interval, což je ideální pro náš scénář. Ale v jiných případech může být výhodné spouštět DAG pouze v pracovních dnech nebo pouze o víkendech.

Použití tagů pro DAGy je novinkou v Airflow a pomáhá s organizováním a kategorizováním DAGů podle týmů nebo specifických funkcí. To může výrazně usnadnit práci, zejména při správě velkých projektů s mnoha DAGy.

Definování výchozích argumentů pro DAG je nezbytné pro správné fungování pracovního toku. dag_owner označuje, kdo je zodpovědný za správu DAGu, což je důležité pro správu a aktualizaci úkolů v rámci týmu. Tento parametr by měl být vždy aktuální a v souladu s tím, kdo má daný DAG na starosti.

Jak vytvořit vlastní plugin pro metrics dashboard v Apache Airflow?

Prvním krokem je integrace knihovny Chart.js, kterou načítáme z open-source CDN. Tato knihovna nám umožňuje vykreslovat grafy do elementů canvas na webové stránce. Na backendu definujeme proměnnou pro data ve formátu JSON, která obsahuje informace o statistice běhů DAGs. Posledním krokem je volání metody new Chart(), která jako první argument přijímá elementy canvas a jako druhý konfiguraci grafu.

Po nastavení potřebných prvků na straně frontendové aplikace přichází krok implementace pluginu do samotného Airflow. Tento krok zahrnuje registraci Flask Blueprint a třídy MetricsDashboardView v Airflow. K tomu je potřeba upravit soubor metrics_plugin/__init__.py, kde je uveden základní kód pro registraci pluginu.

python
Copy code
from airflow.plugins_manager import AirflowPlugin

from flask import Blueprint
from plugins.metrics_plugin.views.dashboard import MetricsDashboardView
metrics_blueprint = Blueprint(
    "Metrics", 
    __name__, 
    template_folder="templates", 
    static_folder="static", 
    static_url_path="/static"
)
class MetricsPlugin(AirflowPlugin):
    name = "Metrics Dashboard Plugin"
    flask_blueprints = [metrics_blueprint]
    appbuilder_views = [{
        "name": "Dashboard", 
        "category": "Metrics", 
        "view": MetricsDashboardView()
    }]

V tomto kódu definujeme Flask Blueprint pro plugin, který slouží jako rozhraní pro zobrazení dashboardu, a třídu MetricsPlugin, která se odvozuje od základní třídy AirflowPlugin. Tato třída zajišťuje integraci pluginu do prostředí Airflow a připojuje potřebné webové stránky do menu Airflow pod kategorií "Metrics" a položkou "Dashboard".

Po implementaci tohoto pluginu se zobrazí grafy, které ukazují statistiky úspěšných a neúspěšných běhů DAGs. Například, graf úspěšných běhů zobrazuje počet běhů, které byly úspěšně dokončeny za poslední den, týden a měsíc. Další graf ukazuje neúspěšné běhy za stejné časové období.

Nyní máme kompletní metrics dashboard. Tento dashboard poskytuje užitečné vizualizace výkonu vašeho prostředí Airflow a umožňuje rychlou identifikaci problémových oblastí, jako jsou neúspěšné DAG běhy. Představte si, že každá z těchto vizualizací může být dále přizpůsobena dle potřeb konkrétního pracovního postupu. Možnosti pro experimentování s různými metrikami a vizualizacemi jsou rozsáhlé a lze je snadno implementovat přidáním nových grafů a statistik.

Při vytváření pluginu je důležité mít na paměti nejen implementaci základního grafu, ale také to, jak budete zobrazovat data z různých časových rámců (například za poslední den, týden nebo měsíc), a jak budete tento dashboard přizpůsobovat pro různé potřeby vašich týmů nebo organizace. Vytvoření vlastního metrics dashboardu v Airflow vám umožní přizpůsobit způsob monitorování výkonu a operací vašeho systému podle specifických požadavků.

Co byste měli vědět při práci s těmito dashboardy

Je nezbytné mít na paměti, že správná implementace pluginu závisí na tom, jak efektivně dokážete pracovat s databází, která poskytuje potřebná data pro vaše grafy. V tomto případě, kdy se zobrazují informace o bězích DAGs, budete potřebovat efektivní způsob dotazování na tuto databázi, aby byly grafy vždy aktuální. Zároveň si dejte pozor na optimalizaci výkonu, zejména pokud máte velké množství dat, protože vizualizace může zpomalit vaše webové rozhraní, pokud není správně nakonfigurováno.