🥇Apache Airflow: rendiamo l'ETL più semplice

Ciao, sono Dmitrij Logvinenko — Data Engineer del dipartimento di analisi del gruppo di aziende «Vezёт».

Vi parlerò di uno strumento straordinario per lo sviluppo di processi ETL — Apache Airflow. Ma Airflow è così versatile e multifunzionale che dovreste considerarlo anche se non vi occupate di flussi di dati, ma avete bisogno di avviare periodicamente alcuni processi e monitorarne l'esecuzione.

E sì, non solo parlerò, ma mostrerò anche: ci sono molti codici, schermate e raccomandazioni nel programma.

Cosa si vede di solito quando si cerca la parola Airflow / Wikimedia Commons

Indice

Introduzione
Parte principale, pratica (e un po' teorica)
Parte finale, informativa e di riferimento
- Rocce che abbiamo raccolto per voi
- Strumenti di ulteriore automazione
Link

Introduzione

Apache Airflow — è proprio come Django:

scritto in Python,
ha un'ottima interfaccia di amministrazione,
è illimitatamente espandibile,

— solo migliore, ed è fatto per scopi completamente diversi, a sapere (come scritto fino al kata):

esecuzione e monitoraggio di compiti su un numero illimitato di macchine (quante ve ne permettono Celery/Kubernetes e la vostra coscienza)
con generazione dinamica di workflow da codice Python molto semplice da scrivere e comprendere
e la possibilità di collegare tra loro qualsiasi database e API utilizzando sia componenti già pronti, sia plugin fatti in casa (che si realizzano in modo estremamente semplice).

Utilizziamo Apache Airflow in questo modo:

raccogliamo dati da fonti diverse (molti istanze di SQL Server e PostgreSQL, vari API con metriche delle applicazioni, persino 1C) in DWH e ODS (per noi sono Vertica e Clickhouse).
come avanzato cron, che avvia processi di consolidamento dei dati su ODS e ne monitora anche la manutenzione.

Fino a poco tempo fa, le nostre esigenze erano soddisfatte da un piccolo server con 32 core e 50 GB di RAM. In Airflow funzionano:

oltre 200 DAG (propriamente workflow, nei quali abbiamo inserito i task),
in ciascuno in media ci sono 70 task,
questi vengono avviati (anche in media) una volta all'ora.

E su come ci siamo espansi, scriverò più sotto, ma ora definiamo l'über-task che risolveremo:

Ci sono tre SQL Server di origine, ognuno con 50 database — istanze di un unico progetto, quindi, la loro struttura è identica (quasi ovunque, muahahah!), e significa che in ciascuno c'è una tabella Ordini (per fortuna, una tabella con questo nome può essere inserita in qualsiasi settore). Raccogliamo i dati, aggiungendo campi ausiliari (server di origine, database di origine, identificatore del task ETL) e, in modo ingenuo, li buttiamo in, diciamo, Vertica.

Andiamo!

Parte principale, pratica (e un po' teorica)

Perché ci serve (e a voi)

Quando gli alberi erano alti e io ero semplice SQL-manipolatore in un rivenditore russo, gestivamo i processi ETL aka flussi di dati con due strumenti a nostra disposizione:

Informatica Power Center — un sistema estremamente complesso, eccezionalmente performante, con il proprio hardware e la sua versione di gestione. Io ho usato, a malapena, l'1% delle sue potenzialità. Perché? Beh, in primo luogo, questa interfaccia, che sembra provenire da un'era passata, ci ha messo sotto pressione in modo psicologico. In secondo luogo, questo aggeggio è progettato per processi estremamente avanzati, un riutilizzo intenso dei componenti e altre importantissime furbate da enterprise. Per quanto riguarda il costo, è come il costo di un'ala di un Airbus A380 all'anno, non ne parliamo.
Attenzione, lo screenshot potrebbe far male a chi ha meno di 30 anni.

SQL Server Integration Server — questo compagno lo abbiamo usato nei nostri flussi interni. Ma davvero: usiamo già SQL Server, e non sfruttare i suoi strumenti ETL sarebbe stata una decisione poco saggia. Tutto in esso è a posto: l'interfaccia è bella, e i report di esecuzione… Ma non è per questo che amiamo i prodotti software, oh no, non è per questo. Possiamo versionare il dtsx (che è fondamentalmente un XML con nodi che si mischiano al salvataggio), ma a che serve? Posso fare un pacchetto di attività che trasferisce cento tabelle da un server a un altro? E cosa dico di mille, vi si staccerebbe il dito indice dal cliccare sul mouse dopo venti. Ma è indubbiamente più alla moda:

Stavamo certamente cercando soluzioni. La questione è arrivata a un generatore SSIS scritto a mano… quasi … e poi ho trovato un nuovo lavoro. E lì mi ha colpito Apache Airflow.

Quando ho scoperto che le descrizioni dei processi ETL sono solo semplice codice Python, non stavo quasi più nella pelle dalla gioia. Così i flussi di dati sono stati versionati e diffusi, e trasferire tabelle con strutture identiche da cento database in un unico target è diventato un compito di codice Python su uno schermo da 13" uno o due volte.

Quando ho scoperto che le descrizioni dei processi ETL sono semplicemente codice Python, stavo per ballare dalla gioia. Ecco come i flussi di dati sono stati sottoposti a versionamento e diff, e trasferire tabelle con una struttura comune da centinaia di database in un unico target è diventata una questione di codice Python su uno schermo da 13 pollici.

Consolidiamo il cluster

Eviterei di creare un asilo infantile qui, e di parlare di cose ovvie, come l'installazione di Airflow, il database che hai scelto, Celery e altre attività descritte nella documentazione.

Affinché possiamo iniziare subito con gli esperimenti, ho abbozzato docker-compose.yml in cui:

Alziamo in pratica Airflow: Scheduler, Webserver. Qui girerà anche Flower per il monitoraggio dei task di Celery (dal momento che è già stato aggiunto in apache/airflow:1.10.10-python3.7, e non ci dispiace);
PostgreSQL, nel quale Airflow scriverà le sue informazioni operative (dati del pianificatore, statistiche di esecuzione, ecc.), e Celery annoterà i task completati;
Redis, che fungerà da broker di task per Celery;
Celery worker, che si occuperà dell'esecuzione effettiva dei task.
Nella cartella ./dags metteremo i nostri file con la descrizione dei DAG. Saranno caricati al volo, quindi non è necessario riavviare l'intero stack dopo ogni piccola modifica.

In alcuni casi il codice negli esempi è fornito incompleto (per non appesantire il testo), e in altri viene modificato nel processo. Puoi vedere esempi di codice interi e funzionanti nel repository. https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

version: '3.4'

x-airflow-config: &airflow-config
  AIRFLOW__CORE__DAGS_FOLDER: /dags
  AIRFLOW__CORE__EXECUTOR: CeleryExecutor
  AIRFLOW__CORE__FERNET_KEY: MJNz36Q8222VOQhBOmBROFrmeSxNOgTCMaVp2_HOtE0=
  AIRFLOW__CORE__HOSTNAME_CALLABLE: airflow.utils.net:get_host_ip_address
  AIRFLOW__CORE__SQL_ALCHEMY_CONN: postgres+psycopg2://airflow:airflow@airflow-db:5432/airflow

  AIRFLOW__CORE__PARALLELISM: 128
  AIRFLOW__CORE__DAG_CONCURRENCY: 16
  AIRFLOW__CORE__MAX_ACTIVE_RUNS_PER_DAG: 4
  AIRFLOW__CORE__LOAD_EXAMPLES: 'False'
  AIRFLOW__CORE__LOAD_DEFAULT_CONNECTIONS: 'False'

  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_RETRY: 'False'
  AIRFLOW__EMAIL__DEFAULT_EMAIL_ON_FAILURE: 'False'

  AIRFLOW__CELERY__BROKER_URL: redis://broker:6379/0
  AIRFLOW__CELERY__RESULT_BACKEND: db+postgresql://airflow:airflow@airflow-db/airflow

x-airflow-base: &airflow-base
  image: apache/airflow:1.10.10-python3.7
  entrypoint: /bin/bash
  restart: always
  volumes:
    - ./dags:/dags
    - ./requirements.txt:/requirements.txt

services:
  # Redis come broker di Celery
  broker:
    image: redis:6.0.5-alpine

  # DB per i metadati di Airflow
  airflow-db:
    image: postgres:10.13-alpine

    environment:
      - POSTGRES_USER=airflow
      - POSTGRES_PASSWORD=airflow
      - POSTGRES_DB=airflow

    volumes:
      - ./db:/var/lib/postgresql/data

  # Contenitore principale con Webserver di Airflow, Scheduler, Celery Flower
  airflow:
    <<: *airflow-base

    environment:
      <
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint initdb &&
          (/entrypoint webserver &) &&
          (/entrypoint flower &) &&
           /entrypoint scheduler"

    ports:
      # Celery Flower
      - 5555:5555
      # Webserver di Airflow
      - 8080:8080

  # Worker di Celery, sarà scalato usando `--scale=n`
  worker:
    <<: *airflow-base

    environment:
      <
      -c " sleep 10 &&
           pip install --user -r /requirements.txt &&
           /entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

Note:

Nella build del composer mi sono basato molto su un'immagine nota puckel/docker-airflow – assicuratene di darci un'occhiata. Potrebbe non servirvi più nulla nella vita.
Tutte le impostazioni di Airflow sono disponibili non solo tramite airflow.cfg, ma anche tramite le variabili d'ambiente (grazie agli sviluppatori), di cui ho approfittato pesantemente.
Naturalmente, non è pronto per la produzione: ho volutamente omesso i heartbeats sui contenitori, non ho fatto attenzione alla sicurezza. Ma ho preparato un minimo adatto per i nostri esperimenti.
Fate attenzione che:
- La cartella con i DAG deve essere accessibile sia dal pianificatore che dai lavoratori.
- Lo stesso vale per tutte le librerie di terze parti: devono essere installate sia sulla macchina con lo scheduler che su quella dei worker.

Ma ora semplicemente:

$ docker-compose up --scale worker=3

Dopo che tutto sarà avviato, potrete accedere alle interfacce web:

Airflow: http://127.0.0.1:8080/admin/
Flower: http://127.0.0.1:5555/dashboard

Concetti di base

Se non hai capito nulla di tutte queste "dag", ecco un breve glossario:

Scheduler — il tipo principale in Airflow, che controlla che siano i robot a lavorare e non gli esseri umani: si occupa del programma, aggiorna i dag, avvia i task.
In effetti, nelle versioni precedenti, aveva problemi di memoria (no, non amnesia, ma perdite) e nei config è persino rimasto un parametro legacy run_duration — l'intervallo per il suo riavvio. Ma ora va tutto bene.
DAG (noto anche come "dag") — "grafo aciclico orientato", ma tale definizione poco dirà a qualcuno, e in sostanza è un contenitore per task che interagiscono tra loro (vedi sotto) o l'equivalente di Package in SSIS e Workflow in Informatica.
Oltre ai dag possono esserci anche subdag, ma probabilmente non ci arriveremo.
DAG Run — un dag inizializzato a cui è assegnato il suo execution_date. Le esecuzioni di uno stesso dag possono tranquillamente funzionare in parallelo (se, naturalmente, hai reso i tuoi task idempotenti).
Operator — sono pezzi di codice responsabili dell'esecuzione di un'azione specifica. Ci sono tre tipi di operatori:
- action, come il nostro amato PythonOperator, che è in grado di eseguire qualsiasi codice Python (valido);
- trasferire, che trasportano dati da un luogo all'altro, ad esempio, MsSqlToHiveTransfer;
- sensor consentirà di reagire o di ritardare l'esecuzione del dag fino al verificarsi di un certo evento. HttpSensor può interpellare l'endpoint specificato, e quando ottiene la risposta giusta, avvia il trasferimento GoogleCloudStorageToS3Operator. Una mente curiosa potrebbe chiedere: "perché? Si possono fare ripetizioni direttamente nell'operatore!" E poi, per non intasare il pool di task con operatori in attesa. Il sensore si attiva, controlla e poi scompare fino al prossimo tentativo.
Task — operatori dichiarati indipendentemente dal tipo e associati al dag vengono elevati al rango di task.
Task instance — quando il generale pianificatore decide che è ora di inviare i task in campo ai lavoratori esecutori (proprio sul posto, se stiamo usando LocalExecutor o su un nodo remoto nel caso di CeleryExecutor), assegna loro un contesto (ossia un insieme di variabili — parametri di esecuzione), espande i modelli di comando o richieste e li colloca in un pool.

Generiamo i task

Innanzitutto, delineiamo lo schema generale del nostro dag, e poi iniziamo a scendere sempre più nel dettaglio, perché stiamo utilizzando alcune soluzioni non banali.

Quindi, in modo semplice, un dag simile apparirà così:

da datetime import timedelta, datetime

da airflow import DAG
da airflow.operators.python_operator import PythonOperator

da commons.datasources import sql_server_ds

dag = DAG('orders',
          schedule_interval=timedelta(hours=6),
          start_date=datetime(2020, 7, 8, 0))

def workflow(**context):
    print(context)

per conn_id, schema in sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

Iniziamo a capire:

Per prima cosa importiamo le librerie necessarie e qualcos'altro;
sql_server_ds è List[namedtuple[str, str]] con i nomi delle connessioni da Airflow Connections e i database da cui recupereremo la nostra tabella;
dag è la dichiarazione del nostro DAG, che deve necessariamente trovarsi in globals(), altrimenti Airflow non lo troverà. Bisogna anche dire al DAG:
- come si chiama orders questo nome apparirà poi nell'interfaccia web,
- che inizierà a lavorare a partire dalla mezzanotte dell'otto luglio,
- e che deve essere eseguito circa ogni 6 ore (per i più esperti, qui al posto di timedelta() è consentita cron-stringa 0 0 0/6 ? * * *, per i meno esperti - un'espressione come @daily);
workflow() farà il lavoro principale, ma non adesso. Ora semplicemente scriveremo il nostro contesto nel log.
E ora una semplice magia per creare i task:
- scorriamo le nostre fonti;
- inizializziamo PythonOperator, che eseguirà la nostra funzione vuota workflow(). Non dimenticate di specificare un nome unico (nel contesto del DAG) per il task e di legarlo al DAG stesso. Il flag provide_context invece fornirà alla funzione argomenti aggiuntivi che raccoglieremo con cura tramite **context.

Per ora è tutto. Cosa abbiamo ottenuto:

un nuovo DAG nell'interfaccia web,
circa cento cinquanta task che verranno eseguiti in parallelo (se le impostazioni di Airflow, Celery e le potenza dei server lo permettono).

Beh, quasi.

Chi si occuperà delle dipendenze?

Per semplificare tutto ciò, ho integrato docker-compose.yml la gestione requirements.txt su tutti i nodi.

Ed ora iniziamo:

I rettangoli grigi sono le istanze dei task, gestite dal pianificatore.

Aspettiamo un po’, i worker prendono in carico i task:

Verdi, ovviamente, sono quelli completati con successo. Rossi - quelli non elaborati con successo.

A proposito, nel nostro ambiente di produzione non c'è nessuna cartella ./dags, sincronizzata tra le macchine - tutti i DAG si trovano in git nel nostro Gitlab, e Gitlab CI distribuisce gli aggiornamenti sui server al momento del merge in master.

Un po' su Flower

Mentre i worker elaborano le nostre funzioni vuote, ricordiamo un altro strumento che può mostrarci qualcosa - Flower.

La prima pagina con informazioni riassuntive sui nodi-worker:

La pagina più affollata con i task che sono stati avviati:

La pagina più noiosa con lo stato del nostro broker:

La pagina più vivace — con grafici dello stato dei task e del loro tempo di esecuzione:

Scarichiamo ciò che non è stato caricato

Quindi, tutti i task sono stati eseguiti, possiamo rimuovere i feriti.

E i feriti erano numerosi, per vari motivi. In caso di utilizzo corretto di Airflow, questi quadrati indicano che i dati non sono sicuramente arrivati.

Dobbiamo controllare il log e riavviare le istanze dei task che sono fallite.

Cliccando su qualsiasi quadrato, vedremo le azioni disponibili:

Possiamo eliminare ciò che è fallito. Cioè, dimentichiamo che c'è stata una rottura, e la stessa istanza del task verrà riprogrammata.

È chiaro che fare così con il mouse su tutti i quadrati rossi non è molto umano — non è questo che ci aspettiamo da Airflow. Naturalmente, abbiamo un'arma di distruzione di massa: Browse/Task Instances

Selezioniamo tutto in una volta e ripristiniamo premendo l'opzione corretta:

Dopo la pulizia, i nostri task appaiono così (stanno già aspettando con impazienza che lo scheduler li programmi):

Connessioni, hook e altre variabili

È tempo di dare un'occhiata al prossimo DAG, update_reports.py:

from collections import namedtuple
from datetime import datetime, timedelta
from textwrap import dedent

from airflow import DAG
from airflow.contrib.operators.vertica_operator import VerticaOperator
from airflow.operators.email_operator import EmailOperator
from airflow.utils.trigger_rule import TriggerRule

from commons.operators import TelegramBotSendMessage

dag = DAG('update_reports',
          start_date=datetime(2020, 6, 7, 6),
          schedule_interval=timedelta(days=1),
          default_args={'retries': 3, 'retry_delay': timedelta(seconds=10)})

Report = namedtuple('Report', 'source target')
reports = [Report(f'{table}_view', table) for table in [
    'reports.city_orders',
    'reports.client_calls',
    'reports.client_rates',
    'reports.daily_orders',
    'reports.order_duration']]

email = EmailOperator(
    task_id='email_success', dag=dag,
    to='{{ var.value.all_the_kings_men }}',
    subject='DWH Reports updated',
    html_content=dedent("""Gentili, i rapporti sono stati aggiornati"""),
    trigger_rule=TriggerRule.ALL_SUCCESS)

tg = TelegramBotSendMessage(
    task_id='telegram_fail', dag=dag,
    tg_bot_conn_id='tg_main',
    chat_id='{{ var.value.failures_chat }}',
    message=dedent("""
         Nataliya, svegliati, abbiamo fatto cadere {{ dag.dag_id }}
        """),
    trigger_rule=TriggerRule.ONE_FAILED)

for source, target in reports:
    queries = [f"TRUNCATE TABLE {target}",
               f"INSERT INTO {target} SELECT * FROM {source}"]

    report_update = VerticaOperator(
        task_id=target.replace('reports.', ''),
        sql=queries, vertica_conn_id='dwh',
        task_concurrency=1, dag=dag)

    report_update >> [email, tg]

Tutti abbiamo mai fatto un aggiornamento di report? Eccola di nuovo: c'è un elenco di fonti da cui prelevare i dati; c'è un elenco di destinazione; non dimentichiamo di segnalare quando tutto è successo o si è rotto (beh, questo non ci riguarda, no).

Facciamo di nuovo un giro nel file e diamo un'occhiata alle nuove cose poco chiare:

from commons.operators import TelegramBotSendMessage — non abbiamo alcun impedimento a creare i nostri operatori, e così abbiamo fatto, realizzando un piccolo wrapper per l'invio di messaggi in Unblocked. (Di questo operatore parleremo ancora più avanti);
default_args={} — un dag può distribuire gli stessi argomenti a tutti i suoi operatori;
to='{{ var.value.all_the_kings_men }}' — campo to sarà dinamicamente creato non hardcoded, utilizzando Jinja e una variabile con un elenco di email che ho diligentemente inserito in Admin/Variables;
trigger_rule=TriggerRule.ALL_SUCCESS — condizione di attivazione dell'operatore. Nel nostro caso, la email verrà inviata ai capi solo se tutte le dipendenze sono state eseguite con successo;
tg_bot_conn_id='tg_main' — argomenti conn_id accettano gli identificativi delle connessioni che creiamo in Admin/Connections;
trigger_rule=TriggerRule.ONE_FAILED — i messaggi su Telegram verranno inviati solo in caso di task falliti;
task_concurrency=1 — impediamo l'esecuzione simultanea di più istanze di task dello stesso task. Altrimenti, otterremo l'esecuzione simultanea di più VerticaOperator (che si riferiscono alla stessa tabella);
report_update >> [email, tg] — tutto VerticaOperator convergerà nell'invio della email e del messaggio, in questo modo:

Ma poiché gli operatori di notifica hanno condizioni di attivazione diverse, funzionerà solo uno di loro. In Tree View appare un po' meno chiaro:

Dirò un paio di parole sui macro e i loro amici — delle variabili.

I macro sono segnaposto Jinja che possono inserire informazioni utili negli argomenti degli operatori. Ad esempio, così:

SELECT
    id,
    payment_dtm,
    payment_type,
    client_id
FROM orders.payments
WHERE
    payment_dtm::DATE = '{{ ds }}'::DATE

{{ ds }} verrà espanso con il contenuto della variabile di contesto execution_date nel formato YYYY-MM-DD: 2020-07-14. La parte migliore è che le variabili di contesto sono fissate a un'istanza particolare del task (un quadratino in Tree View), e al riavvio i segnaposto si espanderanno nei medesimi valori.

I valori assegnati possono essere visualizzati tramite il pulsante Rendered in ciascuna istanza del task. Ecco come appare l'istanza del task con l'invio della email:

Ecco come appare l'istanza del task con l'invio del messaggio:

L'elenco completo dei macro incorporati per l'ultima versione disponibile è disponibile qui: Macros Reference

Inoltre, grazie ai plugin, possiamo dichiarare i nostri macro, ma questa è un'altra storia.

Oltre agli elementi predefiniti, possiamo inserire i valori delle nostre variabili (ho già fatto uso di questo nel codice sopra). Creiamo in Admin/Variables alcuni elementi:

Ecco fatto, possiamo usarlo:

TelegramBotSendMessage(chat_id='{{ var.value.failures_chat }}')

Il valore può essere uno scalare, ma può anche essere un JSON. In caso di JSON:

bot_config

{
    "bot": {
        "token": 881hskdfASDA16641,
        "name": "Verter"
    },
    "service": "TG"
}

basta usare il percorso della chiave desiderata: {{ var.json.bot_config.bot.token }}.

Dirò solo una parola e mostrerò uno screenshot riguardante le connessioni. Qui tutto è elementare: nella pagina Admin/Connections creiamo una connessione, inseriamo le nostre credenziali e parametri più specifici. Ecco come:

Le password possono essere crittografate (in modo più accurato rispetto alla versione predefinita), oppure è possibile non specificare il tipo di connessione (come ho fatto per tg_main) — il fatto è che l'elenco dei tipi è hardcoded nei modelli di Airflow e l'estensione non può essere modificata senza toccare il codice sorgente (se per caso non ho trovato qualcosa — chiedetemi di correggermi), ma ottenere le credenziali semplicemente per nome non ci ostacola.

Inoltre, è possibile creare più connessioni con lo stesso nome: in tal caso il metodo BaseHook.get_connection(), che recupera le connessioni per nome, restituirà una qualsiasi di diversi omonimi (sarebbe più logico fare un Round Robin, ma lasciamo questo alla discrezione degli sviluppatori di Airflow).

Variables e Connections sono strumenti davvero utili, ma è importante non perdere l'equilibrio: quali parti dei tuoi flussi conservi effettivamente nel codice e quali affidi a Airflow. Da un lato, cambiare rapidamente un valore, ad esempio l'indirizzo email per la distribuzione, può essere conveniente tramite UI. Dall'altro, tuttavia, si tratta pur sempre di tornare al clic del mouse, da cui noi (io) volevamo scappare.

Lavorare con le connessioni è uno dei compiti hook. In generale, i hook di Airflow sono punti di connessione con servizi e librerie esterne. Ad esempio, JiraHook ci aprirà un client per interagire con Jira (possiamo spostare le attività da un lato all'altro), mentre con SambaHook è possibile caricare un file locale su smb-share.

Analizziamo un operatore personalizzato

E ci stiamo avvicinando a vedere come è fatto il TelegramBotSendMessage

Codice commons/operators.py con l'effettivo operatore:

dalla digitazione importare Unione

dall'airflow.operatori importare BaseOperator

dai commons.hooks importare TelegramBotHook, TelegramBot

classe TelegramBotSendMessage(BaseOperator):
    """Invia un messaggio a chat_id utilizzando TelegramBotHook

    Esempio:
        >>> TelegramBotSendMessage(
        ...     task_id='telegram_fail', dag=dag,
        ...     tg_bot_conn_id='tg_bot_default',
        ...     chat_id='{{ var.value.all_the_young_dudes_chat }}',
        ...     message='{{ dag.dag_id }} failed :(',
        ...     trigger_rule=TriggerRule.ONE_FAILED)
    """
    template_fields = ['chat_id', 'message']

    def __init__(self,
                 chat_id: Unione[int, str],
                 message: str,
                 tg_bot_conn_id: str = 'tg_bot_default',
                 *args, **kwargs):
        super().__init__(*args, **kwargs)

        self._hook = TelegramBotHook(tg_bot_conn_id)
        self.client: TelegramBot = self._hook.client
        self.chat_id = chat_id
        self.message = message

    def execute(self, context):
        print(f'Invio "{self.message}" alla chat {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

Qui, come il resto di Airflow, è tutto molto semplice:

Abbiamo ereditato da BaseOperator, che implementa molte cose specifiche di Airflow (dai un'occhiata quando hai tempo)
Abbiamo dichiarato i campi template_fields, nei quali Jinja cercherà macro per l'elaborazione.
Abbiamo organizzato correttamente gli argomenti per __init__(), impostando i valori predefiniti dove necessario.
Non abbiamo dimenticato l'inizializzazione del genitore.
Abbiamo aperto il hook corrispondente TelegramBotHook, ottenendo da esso l'oggetto client.
Abbiamo sovrascritto (ridefinito) il metodo BaseOperator.execute(), che Airflow chiamerà quando sarà il momento di eseguire l'operatore: qui implementiamo l'azione principale, senza dimenticare di fare logging. (Il logging, tra l'altro, viene effettuato direttamente in stdout e stderr — Airflow gestirà tutto, lo incapsulerà in modo elegante, e lo organizzerà dove necessario.)

Vediamo cosa abbiamo in commons/hooks.py. La prima parte del file, con il hook stesso:

dalla digitazione importare Unione

dall'airflow.hooks.base_hook importare BaseHook
dalla requests_toolbelt.sessions importare BaseUrlSession

classe TelegramBotHook(BaseHook):
    """Hook dell'API del Bot Telegram

    Nota: aggiungi una connessione con Conn Type vuoto e non dimenticare
    di compilare Extra:

        {"bot_token": "YOuRAwEsomeBOtToKen"}
    """
    def __init__(self,
                 tg_bot_conn_id='tg_bot_default'):
        super().__init__(tg_bot_conn_id)

        self.tg_bot_conn_id = tg_bot_conn_id
        self.tg_bot_token = None
        self.client = None
        self.get_conn()

    def get_conn(self):
        extra = self.get_connection(self.tg_bot_conn_id).extra_dejson
        self.tg_bot_token = extra['bot_token']
        self.client = TelegramBot(self.tg_bot_token)
        return self.client

Non so nemmeno cosa si possa spiegare qui, evidenzio solo i punti importanti:

Ereditiamo, pensiamo agli argomenti: nella maggior parte dei casi ce ne sarà uno solo: conn_id;
Sovrascriviamo i metodi standard: io mi sono limitato a get_conn(), nel quale ottengo i parametri di connessione per nome e semplicemente estraggo la sezione extra (questo campo per JSON), in cui ho (secondo la mia stessa istruzione!) inserito il token del bot Telegram: {"bot_token": "YOuRAwEsomeBOtToKen"}.
Creando un'istanza del nostro TelegramBot, passando già un token specifico.

Ecco fatto. È possibile ottenere il cliente dall'hook usando TelegramBotHook().clent o TelegramBotHook().get_conn().

E la seconda parte del file, dove ho realizzato un micro-wrapper per l'API REST di Telegram, per non dover portare lo stesso python-telegram-bot solo per un metodo sendMessage.

class TelegramBot:
    """Wrapper per l'API del Bot di Telegram

    Esempi:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Ciao, tesoro')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Ciao, tesoro', chat_id=-1762374628374)
    """
    API_ENDPOINT = 'https://api.telegram.org/bot{}/'

    def __init__(self, tg_bot_token: str, chat_id: Union[int, str] = None):
        self._base_url = TelegramBot.API_ENDPOINT.format(tg_bot_token)
        self.session = BaseUrlSession(self._base_url)
        self.chat_id = chat_id

    def send_message(self, message: str, chat_id: Union[int, str] = None):
        method = 'sendMessage'

        payload = {'chat_id': chat_id or self.chat_id,
                   'text': message,
                   'parse_mode': 'MarkdownV2'}

        response = self.session.post(method, data=payload).json()
        if not response.get('ok'):
            raise TelegramBotException(response)

class TelegramBotException(Exception):
    def __init__(self, *args, **kwargs):
        super().__init__((args, kwargs))

La strada giusta è comporre tutto questo: TelegramBotSendMessage, TelegramBotHook, TelegramBot — in un plugin, mettere in un repository pubblico e fornire come Open Source.

Mentre studiavamo tutto questo, i nostri aggiornamenti delle segnalazioni erano riusciti a accumularsi e a inviarmi un messaggio di errore nel canale. Vado a controllare cosa non va di nuovo...

Qualcosa si è rotto nel nostro DAG! Dobbiamo aspettarci proprio questo, giusto?

Bevi qualcosa?

Hai la sensazione che io abbia saltato qualcosa? Sembrava che avessi promesso di riversare i dati da SQL Server a Vertica, e qui sono uscito dal tema, che disgraziato!

Questo atto era intenzionale, dovevo solo spiegarti un po' di terminologia. Ora possiamo andare avanti.

Il nostro piano era il seguente:

Creare il DAG
Generare i task
Vedere come tutto appare bello
Assegnare i numeri di sessione agli upload
Prelevare i dati da SQL Server
Inserire i dati in Vertica
Raccogliere statistiche

Quindi, per avviare tutto questo, ho fatto un piccolo supplemento al nostro docker-compose.yml:

docker-compose.db.yml

version: '3.4'

x-mssql-base: &mssql-base
  image: mcr.microsoft.com/mssql/server:2017-CU21-ubuntu-16.04
  restart: always
  environment:
    ACCEPT_EULA: Y
    MSSQL_PID: Express
    SA_PASSWORD: SayThanksToSatiaAt2020
    MSSQL_MEMORY_LIMIT_MB: 1024

services:
  dwh:
    image: jbfavre/vertica:9.2.0-7_ubuntu-16.04

  mssql_0:
    <<: *mssql-base

  mssql_1:
    <<: *mssql-base

  mssql_2:
    <<: *mssql-base

  mssql_init:
    image: mio101/py3-sql-db-client-base
    command: python3 ./mssql_init.py
    depends_on:
      - mssql_0
      - mssql_1
      - mssql_2
    environment:
      SA_PASSWORD: SayThanksToSatiaAt2020
    volumes:
      - ./mssql_init.py:/mssql_init.py
      - ./dags/commons/datasources.py:/commons/datasources.py

Lì alziamo:

Vertica come host dwh con le impostazioni predefinite,
tre istanze di SQL Server,
popoliamo i database con alcune ultime informazioni (non guardate mai nel mssql_init.py!)

Avviamo tutto con un comando leggermente più complesso rispetto all'ultima volta:

$ docker-compose -f docker-compose.yml -f docker-compose.db.yml up --scale worker=3

Quello che ha generato il nostro meraviglioso randomizzatore può essere visualizzato utilizzando l'opzione Data Profiling/Ad Hoc Query:

Cosa più importante, non mostrarlo agli analisti

Non mi soffermerò dettagliatamente sulle sessioni ETL non lo farò, è tutto banale: creiamo un database, all'interno una tabella, incapsuliamo tutto in un gestore di contesto e ora facciamo così:

with Session(task_name) as session:
    print('Load', session.id, 'started')

    # Load workflow
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

dalla sys importa stderr

class Session:
    """Sessione del flusso di lavoro ETL

    Esempio:
        con Session(task_name) come session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Ben fatto'
    """

    def __init__(self, connection, task_name):
        self.connection = connection
        self.connection.autocommit = True

        self._task_name = task_name
        self._id = None

        self.loaded_rows = None
        self.successful = None
        self.comment = None

    def __enter__(self):
        return self.open()

    def __exit__(self, exc_type, exc_val, exc_tb):
        if any(exc_type, exc_val, exc_tb):
            self.successful = False
            self.comment = f'{exc_type}: {exc_val}n{exc_tb}'
            print(exc_type, exc_val, exc_tb, file=stderr)
        self.close()

    def __repr__(self):
        return (f'')

    @property
    def task_name(self):
        return self._task_name

    @property
    def id(self):
        return self._id

    def _execute(self, query, *args):
        with self.connection.cursor() as cursor:
            cursor.execute(query, args)
            return cursor.fetchone()[0]

    def _create(self):
        query = """
            CREATE TABLE IF NOT EXISTS sessions (
                id          SERIAL       NOT NULL PRIMARY KEY,
                task_name   VARCHAR(200) NOT NULL,

                started     TIMESTAMPTZ  NOT NULL DEFAULT current_timestamp,
                finished    TIMESTAMPTZ           DEFAULT current_timestamp,
                successful  BOOL,

                loaded_rows INT,
                comment     VARCHAR(500)
            );
            """
        self._execute(query)

    def open(self):
        query = """
            INSERT INTO sessions (task_name, finished)
            VALUES (%s, NULL)
            RETURNING id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'aperta')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Sessione non aperta')
        query = """
            UPDATE sessions
            SET
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            WHERE
                id = %s
            RETURNING id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'chiusa',
              ', successo: ', self.successful,
              ', Caricati: ', self.loaded_rows,
              ', commento:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

È arrivato il momento di recuperare i nostri dati dalle nostre centocinquanta tabelle. Lo faremo con delle righe molto semplici:

source_conn = MsSqlHook(mssql_conn_id=src_conn_id, schema=src_schema).get_conn()

query = f"""
    SELECT 
        id, start_time, end_time, type, data
    FROM dbo.Orders
    WHERE
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)

Con l'ausilio del hook otteniamo da Airflow pymssql-connessione
Nel query inseriamo un vincolo sotto forma di data — il templater lo passerà alla funzione.
Riforniamo la nostra query pandas, che estrarrà per noi DataFrame — ci sarà utile in seguito.

Utilizzo la sostituzione {dt} al posto del parametro di query %s non perché sia un Burattino malvagio, ma perché pandas non può gestire pymssql e lo sottopone all'ultimo params: List, anche se lui desidera molto la tupla.
Si prega di notare che lo sviluppatore pymssql ha deciso di non supportarlo più, ed è ora di passare a pyodbc.

Vediamo come Airflow ha riempito gli argomenti delle nostre funzioni:

Se non ci sono dati, non ha senso continuare. Ma è anche strano considerare il caricamento come riuscito. Ma non è un errore. A-a-a, cosa fare?! Ecco cosa:

if df.empty:
    raise AirflowSkipException('Nessuna riga da caricare')

AirflowSkipException dirà ad Airflow che in realtà non c'è un errore, e che saltiamo il task. Nell'interfaccia non ci sarà un quadratino verde o rosso, ma di colore rosa.

Aggiungiamo ai nostri dati alcune colonne:

df['etl_source'] = src_schema
df['etl_id'] = session.id
df['hash_id'] = hash_pandas_object(df[['etl_source', 'id']])

Cioè:

Il database da cui abbiamo prelevato gli ordini,
L'identificatore della nostra sessione di caricamento (sarà diverso per ogni task),
L'hash dalla fonte e dall'identificatore dell'ordine — in modo che nella base finale (dove tutto viene raccolto in un'unica tabella) abbiamo un identificatore univoco dell'ordine.

Rimane il penultimo passo: caricare tutto in Vertica. E, strano ma vero, uno dei modi più efficaci per farlo è tramite CSV!

# Export data to CSV buffer
buffer = StringIO()
df.to_csv(buffer,
          index=False, sep='|', na_rep='NUL', quoting=csv.QUOTE_MINIMAL,
          header=False, float_format='%.8f', doublequote=False, escapechar='\')
buffer.seek(0)

# Push CSV
target_conn = VerticaHook(vertica_conn_id=target_conn_id).get_conn()

copy_stmt = f"""
    COPY {target_table}({df.columns.to_list()}) 
    FROM STDIN 
    DELIMITER '|' 
    ENCLOSED '"' 
    ABORT ON ERROR 
    NULL 'NUL'
    """

cursor = target_conn.cursor()
cursor.copy(copy_stmt, buffer)

Stiamo creando un ricevitore speciale StringIO.
pandas si occuperà gentilmente di incapsulare il nostro DataFrame come CSV-stringhe.
Apriremo una connessione al nostro amato Vertica tramite hook.
E ora, con l'aiuto di copy() invieremo i nostri dati direttamente a Vertica!

Dal driver recupereremo quante righe sono state caricate e diremo al manager della sessione che va tutto bene:

session.loaded_rows = cursor.rowcount
session.successful = True

Ecco tutto.

In produzione, creiamo manualmente la tabella di destinazione. Qui, mi sono concesso un piccolo automatismo:

create_schema_query = f'CREATE SCHEMA IF NOT EXISTS {target_schema};'
create_table_query = f"""
    CREATE TABLE IF NOT EXISTS {target_schema}.{target_table} (
         id         INT,
         start_time TIMESTAMP,
         end_time   TIMESTAMP,
         type       INT,
         data       VARCHAR(32),
         etl_source VARCHAR(200),
         etl_id     INT,
         hash_id    INT PRIMARY KEY
     );"""

create_table = VerticaOperator(
    task_id='create_target',
    sql=[create_schema_query,
         create_table_query],
    vertica_conn_id=target_conn_id,
    task_concurrency=1,
    dag=dag)

Con l'aiuto di VerticaOperator() creo lo schema del DB e la tabella (se non ci sono già, ovviamente). La cosa importante è posizionare correttamente le dipendenze:

for conn_id, schema in sql_server_ds:
    load = PythonOperator(
        task_id=schema,
        python_callable=workflow,
        op_kwargs={
            'src_conn_id': conn_id,
            'src_schema': schema,
            'dt': '{{ ds }}',
            'target_conn_id': target_conn_id,
            'target_table': f'{target_schema}.{target_table}'},
        dag=dag)

    create_table >> load

Riepiloghiamo

— Ecco, — disse il topino, — non è vero che adesso
Sei sicuro che nella foresta io sia la creatura più spaventosa?

Julia Donaldson, «Il Gruffalò»

Penso che se io e i miei colleghi facessimo una competizione: chi crea e avvia più rapidamente un processo ETL da zero: loro con i loro SSIS e il mouse e io con Airflow… E poi confronteremmo anche la facilità di gestione… Uff, credo che sarete d'accordo, io li supererei su tutti i fronti!

Se vogliamo essere un po' più seri, Apache Airflow — grazie alla descrizione dei processi sotto forma di codice — ha reso il mio lavoro fare molto di più comodo e piacevole.

La sua illimitata espandibilità: sia in termini di plugin che di predisposizione alla scalabilità — vi permette di utilizzare Airflow praticamente in qualsiasi ambito: sia che si tratti di un ciclo completo di raccolta, preparazione e elaborazione dei dati, sia nel lancio di razzi (verso Marte, ovviamente).

Parte finale, informativa e di riferimento

Rocce che abbiamo raccolto per voi

start_date. Sì, questo è già un meme locale. Attraverso il parametro principale del dag start_date passano tutti. In breve, se si indica in start_date la data attuale e in schedule_interval — un giorno, allora il DAG si avvierà domani non prima.
```
start_date = datetime(2020, 7, 7, 0, 1, 2)
```
E non ci sono più problemi.
Ad essa è collegato anche un altro errore di esecuzione: Il task manca del parametro start_date, che di solito indica che hai dimenticato di associare il DAG all'operatore.
Tutto su una sola macchina. Sì, e il database (di Airflow stesso e della nostra interfaccia), il server web, il pianificatore e i worker. E funzionava anche. Ma col tempo il numero di task nei servizi è aumentato, e quando PostgreSQL ha cominciato a rispondere per indice in 20 ms invece di 5, lo abbiamo preso e ci siamo spostati.
LocalExecutor. Sì, siamo ancora su di esso, e siamo già arrivati al limite. Il LocalExecutor ci è sempre bastato, ma ora è giunto il momento di espanderci di almeno un worker, e dovremo impegnarci per passare a CeleryExecutor. E poiché con esso si può lavorare anche su una sola macchina, nulla ci impedisce di usare Celery anche su un server che "naturalmente, non andrà mai in produzione, lo giuro!"
Non utilizzare strumenti integrati:
- Connections per conservare le credenziali dei servizi,
- SLA Misses per reagire a task che non sono stati completati in tempo,
- XCom per lo scambio di metadati (ho detto metadati!) tra i task del dag.dati!) tra i task del dag.
Abuso della posta. Cosa dire? Avevo impostato notifiche per tutti i ripetuti compiti non riusciti. Ora nella mia Gmail di lavoro ho più di 90k email da Airflow e l'interfaccia web della posta si rifiuta di prendere e cancellare più di 100 alla volta.

Altri insidie: Apache Airflow Pitfails

Strumenti di ulteriore automazione

Per farci lavorare di più con la testa e non con le mani, Airflow ci ha preparato questo:

REST API — ha ancora lo status Experimental, il che non gli impedisce di funzionare. Con questo strumento è possibile non solo ricevere informazioni sui DAG e sui compiti, ma anche fermare/avviare un DAG, creare un DAG Run o un pool.
CLI — dalla riga di comando sono disponibili molti strumenti che non solo non sono pratici da usare tramite WebUI, ma che sono addirittura assenti. Ad esempio:
- backfill è necessario per riavviare le istanze dei compiti.
  Per esempio, se arrivano gli analisti e dicono: «Signore, ci sono problemi nei dati dal 1 al 13 gennaio! Ripara-ripara-ripara-ripara!». E tu fai:
```
airflow backfill -s '2020-01-01' -e '2020-01-13' orders
```
- Manutenzione del database: initdb, resetdb, upgradedb, checkdb.
- run, che permette di avviare un'istanza di un compito, tralasciando tutte le sue dipendenze. Inoltre, è possibile eseguirlo tramite LocalExecutor, anche se hai un cluster Celery.
- Fa più o meno la stessa cosa test, solo che non scrive nulla nel database.
- connections permette di creare in massa connessioni dalla shell.

Python API è un modo piuttosto hardcore di interagire, pensato per i plugin e non per farsi le cose a mano. Ma chi ci impedisce di andare in /home/airflow/dags, lanciare ipython e iniziare a fare ciò che vogliamo? Ad esempio, si può esportare tutte le connessioni con questo codice:

from airflow import settings
from airflow.models import Connection

fields = 'conn_id conn_type host port schema login password extra'.split()

session = settings.Session()
for conn in session.query(Connection).order_by(Connection.conn_id):
  d = {field: getattr(conn, field) for field in fields}
  print(conn.conn_id, '=', d)

Connessione al database dei metadati di Airflow. Non consiglio di scriverci, ma estrarre lo stato dei compiti per varie metriche specifiche può essere fatto in modo molto più veloce e semplice rispetto a qualsiasi API.

Diciamo che non tutti i nostri compiti sono idempotenti, e a volte possono fallire, ed è normale. Ma diversi fallimenti sono già preoccupanti, e bisognerebbe controllare.

Attenzione, SQL!

CON LE ultime esecuzioni COME (
SELEZIONA
    task_id,
    dag_id,
    execution_date,
    stato,
        row_number()
        OVER (
            PARTIZIONE PER task_id, dag_id
            ORDINARE PER execution_date DESC) COME rn
DA public.task_instance
DOVE
    execution_date > now() - INTERVAL '2' GIORNI
),
fallite COME (
    SELEZIONA
        task_id,
        dag_id,
        execution_date,
        stato,
        CASE WHEN rn = row_number() OVER (
            PARTIZIONE PER task_id, dag_id
            ORDINARE PER execution_date DESC)
                 ALLORA VERO FINE AS last_fail_seq
    DA last_executions
    DOVE
        stato IN ('failed', 'up_for_retry')
)
SELEZIONA
    task_id,
    dag_id,
    count(last_fail_seq)                       COME unsuccessful,
    count(CASE WHEN last_fail_seq
        E stato = 'failed' ALLORA 1 FINE)       COME failed,
    count(CASE WHEN last_fail_seq
        E stato = 'up_for_retry' ALLORA 1 FINE) COME up_for_retry
DA fallite
GROUP BY
    task_id,
    dag_id
HAVING
    count(last_fail_seq) > 0

Link

E ovviamente, i primi dieci collegamenti nei risultati di Google contengono il contenuto della cartella Airflow dai miei preferiti.

Documentazione di Apache Airflow — naturalmente, è necessario iniziare con la documentazione ufficiale, ma chi legge le istruzioni?
Migliori Pratiche — almeno leggete le raccomandazioni degli sviluppatori.
L'interfaccia utente di Airflow — l'inizio: interfaccia utente a immagini
Comprendere i concetti chiave di Apache Airflow — spiega bene i concetti di base, nel caso (chissà mai!) non abbiate capito qualcosa da me.
Blog di Tianlong — Una guida su come costruire un server/cluster Airflow — una guida breve per la configurazione di un cluster Airflow.
Esecuzione di Apache Airflow presso Lyft — un articolo quasi altrettanto interessante, solo un po' più formale, con meno esempi.
Come Apache Airflow distribuisce i lavori sui worker di Celery — sul lavoro in combinazione con Celery.
Migliori pratiche di scrittura per DAG in Apache Airflow — riguardo all'idempotenza dei task, al caricamento per ID invece che per data, alle trasformazioni, alla struttura dei file e altre cose interessanti.
Gestire le dipendenze in Apache Airflow — dipendenze dei task e Regole di Attivazione, che ho menzionato solo di sfuggita.
Airflow: Quando il tuo DAG è molto indietro rispetto al programma — come superare alcune situazioni in cui "funziona come previsto" nel pianificatore, caricare dati mancanti e stabilire le priorità dei task.
Query SQL utili per Apache Airflow — utili query SQL per i metadati di Airflow.
Inizia a sviluppare flussi di lavoro con Apache Airflow — c'è una sezione utile sulla creazione di un sensore personalizzato.
Costruire l'infrastruttura di Data Science Fetchr su AWS con Presto e Airflow — una breve nota interessante sulla costruzione di un'infrastruttura su AWS per Data Science.
7 errori comuni da controllare quando si effettua il debug dei DAG di Airflow — errori comuni (quando qualcuno non legge le istruzioni).
Memorizzare e accedere a password utilizzando Apache Airflow — sorridi, osserva come le persone aggirano la memorizzazione delle password, quando basterebbe usare semplicemente le Connections.
Il Zen di Python e Apache Airflow — passaggio implicito del DAG, passaggio del contesto nella funzione, di nuovo sulle dipendenze, e anche sul salto delle esecuzioni dei task.
Airflow: Consigli, Trucchi e Pratiche Meno Conosciute — sull'uso di argomenti predefiniti e params nei modelli, così come su Variabili e Connessioni.
Profiling del Pianificatore Airflow — raccontando come il pianificatore è stato preparato per Airflow 2.0.
Apache Airflow con 3 lavoratori Celery in docker-compose — un articolo un po' datato sul deployment del nostro cluster in docker-compose.
4 Task di Template Utilizzando il Contesto di Airflow — task dinamici tramite modelli e passaggio di contesto.
Notifiche di Errore in Airflow — avvisi standard e personalizzati via email e Slack.
Workshop Airflow: DAG complessi senza workaround — ramificazioni dei task, macro e XCom.