ProHoster > Blogg > Administration > Apache Airflow: gör ETL enklare

Apache Airflow: gör ETL enklare

Hej, jag heter Dmitrij Logvinenko — Data Engineer pĂ„ analysavdelningen inom gruppen av företag "Vezot".

Jag kommer att berĂ€tta för er om ett fantastiskt verktyg för att utveckla ETL-processer — Apache Airflow. Men Airflow Ă€r sĂ„ mĂ„ngsidigt och komplett att ni borde ta en nĂ€rmare titt pĂ„ det Ă€ven om ni inte arbetar med dataflöden, utan har behov av att periodiskt starta vissa processer och övervaka deras genomförande.

Och ja, jag kommer inte bara att berÀtta utan ocksÄ visa: programmet innehÄller mycket kod, skÀrmbilder och rekommendationer.

Apache Airflow: gör ETL enklare
Vad ser man vanligtvis nÀr man googlar ordet Airflow / Wikimedia Commons

InnehÄllsförteckning

Introduktion

Apache Airflow — det Ă€r precis som Django:

  • skrivet i Python,
  • har ett utmĂ€rkt administrationsgrĂ€nssnitt,
  • obegrĂ€nsat utbyggbart,

— bara bĂ€ttre, och det Ă€r helt gjort för andra Ă€ndamĂ„l, nĂ€mligen (som det stĂ„r skrivet innan koden):

  • start och övervakning av uppgifter pĂ„ ett obegrĂ€nsat antal maskiner (sĂ„ mĂ„nga som Celery/Kubernetes och ditt samvete tillĂ„ter)
  • med dynamisk generering av arbetsflöden frĂ„n mycket lĂ€tt skriven och förstĂ„elig Python-kod
  • och möjligheten att lĂ€nka ihop nĂ€stan alla databaser och API:er med hjĂ€lp av bĂ„de fĂ€rdiga komponenter och egna plugins (vilket görs extremt enkelt).

Vi anvÀnder Apache Airflow sÄ hÀr:

  • vi samlar data frĂ„n olika kĂ€llor (mĂ„nga instanser av SQL Server och PostgreSQL, olika API:er med app-metrik, till och med 1C) till DWH och ODS (hos oss Ă€r det Vertica och Clickhouse).
  • som en avancerad cron, som startar konsolideringsprocesser pĂ„ ODS och övervakar deras underhĂ„ll.

Tills nyligen tÀcktes vÄra behov av en liten server med 32 kÀrnor och 50 GB RAM. I Airflow fungerar:

  • över 200 DAG-ar (faktiska arbetsflöden, dĂ€r vi har fyllt uppgifter),
  • i genomsnitt med 70 uppgifter,
  • detta körs (ocksĂ„ i genomsnitt) varje timme.

Och om hur vi expanderade kommer jag att skriva nedan, men nu lÄt oss definiera den övergripande uppgiften vi kommer att lösa:

Det finns tre ursprungliga SQL-servrar, var och en med 50 databaser - instanser av samma projekt, sÄ strukturen Àr densamma (nÀstan överallt, muahaha), vilket innebÀr att varje server har en tabell som heter Orders (tack och lov kan vi trycka in en tabell med det namnet i vilket företag som helst). Vi hÀmtar data och lÀgger till hjÀlpfÀlt (kÀlla-server, kÀlla-databas, ETL-uppgiftsidentifierare) och naivt slÀnger vi dem i, sÀg, Vertica.

LÄt oss köra!

Huvuddelen, praktisk (och lite teoretisk)

Varför behöver vi det (och ni)

NÀr trÀden var stora och jag var enkel SQL-arbetare i en rysk detaljhandel, körde vi ETL-processer aka datastreams med hjÀlp av tvÄ verktyg som vi hade tillgÀngliga:

  • Informatica Power Center — ett mycket omfattande system, extremt kraftfullt, med sin egen hĂ„rdvara och egen versionering. Jag anvĂ€nde kanske 1 % av dess kapabiliteter. Varför? Jo, för det första, det hĂ€r grĂ€nssnittet var psykologiskt pressande och kĂ€ndes kvar i 00-talet. För det andra, den hĂ€r grejen Ă€r designad för extremt avancerade processer, intensiv Ă„teranvĂ€ndning av komponenter och andra mycket viktiga företags-finesser. Vi ska vara tyst om att det kostar som en Airbus A380 per Ă„r.

    Försiktighet, en skÀrmdump kan göra ont för yngre Àn 30 Är

    Apache Airflow: gör ETL enklare

  • SQL Server Integration Server — detta verktyg anvĂ€nde vi i vĂ„rt interna projektdatastream. Men Ă€rligt talat: vi anvĂ€nder redan SQL Server, och att inte anvĂ€nda dess ETL-verktyg skulle vara irrationellt. Allt i det Ă€r bra: grĂ€nssnittet Ă€r snyggt, och det finns rapporter om utförande... Men vi Ă€lskar inte programvaruprodukter för det, oh nej. Vi kan versionera den dtsx (som Ă€r en XML med blandade noder vid sparande) men vad hjĂ€lper det? Att skapa ett paket med uppgifter som kan flytta hundra tabeller frĂ„n en server till en annan? Nej, vad hundra, din pekfinger kommer att falla av efter att ha klickat pĂ„ musknappen tjugo gĂ„nger. Men det ser definitivt mer modernt ut:

    Apache Airflow: gör ETL enklare

Vi letade definitivt efter lösningar. Faktum Ă€r att nĂ€stan det gick sĂ„ lĂ„ngt att skriva en egen generator för SSIS-paket



 men sedan fick jag nytt jobb. Och dĂ€r stötte jag pĂ„ Apache Airflow.

NÀr jag fick veta att beskrivningar av ETL-processer Àr enkel Python-kod, nÀstan dansade jag av glÀdje. SÄ hÀr fick datastreams versionshantering och diff, och att slÄ samman tabeller med en enhetlig struktur frÄn hundra databaser till en destination blev en frÄga om Python-kod pÄ en och en halv till tvÄ 13-tums skÀrmar.

Bygger en kluster

LÄt oss inte ha en barnslig diskussion och prata om sjÀlvklara saker som installation av Airflow, den databas du valt, Celery och andra frÄgor som beskrivs i dokumentationen.

För att vi ska kunna sÀtta igÄng med experimenten genast, har jag skissat pÄ docker-compose.yml dÀr:

  • LĂ„t oss starta Airflow: Scheduler, Webserver. DĂ€r kommer Flower att köras för att övervaka Celery-uppgifter (eftersom det redan har inkluderats i apache/airflow:1.10.10-python3.7, och vi Ă€r inte emot det);
  • PostgreSQL, dĂ€r Airflow kommer att skriva sin logginformation (schemalĂ€ggarens data, utförandestatistik osv.), och Celery – notera avslutade uppgifter;
  • Redis, som kommer att fungera som uppgiftsbrygga för Celery;
  • Celery-arbetare, som kommer att utföra uppgifterna direkt.
  • I mappen .\/dags vi kommer att lĂ€gga vĂ„ra filer med beskrivning av dags. De kommer att fĂ„ngas upp i realtid, sĂ„ det Ă€r inte nödvĂ€ndigt att dra om hela stacken efter varje litet fel.

I vissa fall Àr koden i exemplen inte helt Ätergiven (för att inte överbelasta texten), och i andra fall modifieras den under processen. Hela fungerande kodexempel kan ses i repositoryt. 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 as a Celery broker
  broker:
    image: redis:6.0.5-alpine

  # DB for the Airflow metadata
  airflow-db:
    image: postgres:10.13-alpine

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

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

  # Main container with Airflow Webserver, 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
      # Airflow Webserver
      - 8080:8080

  # Celery worker, will be scaled using `--scale=n`
  worker:
    <<: *airflow-base

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

    depends_on:
      - airflow
      - airflow-db
      - broker

Noteringar:

  • I built the compose file mainly relying on the well-known image puckel/docker-airflow — definitely check it out. You might not need anything else in your life.
  • All Airflow settings are available not only through airflow.cfg, but also via environment variables (thanks to the developers), which I took full advantage of.
  • Of course, it's not production-ready: I intentionally did not set heartbeats on the containers and did not bother with security. But I created a minimum setup suitable for our experiments.
  • Note that:
    • The folder with the DAGs must be accessible to both the scheduler and the workers.
    • The same applies to all third-party libraries—they must all be installed on the machines with the scheduler and the workers.

Now simply run:

$ docker-compose up --scale worker=3

Once everything is up, you can check the web interfaces:

GrundlÀggande begrepp

Om du inte förstod nÄgot av alla dessa "dags", hÀr Àr en kort ordbok:

  • SchemalĂ€ggare — den viktigaste personen i Airflow, som sĂ€kerstĂ€ller att robotar jobbar och inte mĂ€nniskor: han övervakar schemat, uppdaterar dags, och startar uppgifter.

    Faktum Ă€r att han hade minnesproblem i Ă€ldre versioner (nej, inte amnesi, utan lĂ€ckor) och det finns till och med en legacy-parameter kvar i konfigurationerna. run_duration — tidsintervallet för hans omstart. Men nu fungerar allt bra.

  • DAG (Ă€ven kallad "dag") — "direkt acyklisk graf", men den definitionen sĂ€ger knappast nĂ„got för de flesta, i grunden Ă€r det en behĂ„llare för uppgifter som interagerar med varandra (se nedan), eller en liknelse till Package i SSIS och Workflow i Informatica.

    Förutom dags kan det ocksÄ finnas subdags, men vi kommer sannolikt inte att nÄ dem.

  • DAG Run — en initialiserad dag som tilldelats sitt execution_date. DAG-körningar av samma dag kan mycket vĂ€l arbeta parallellt (om du förstĂ„s har gjort dina uppgifter idempotenta).
  • Operatör — bitarna av kod som ansvarar för att utföra en specifik Ă„tgĂ€rd. Det finns tre typer av operatörer:
    • Ă„tgĂ€rd, som till exempel vĂ„r favorit, PythonOperator, som kan utföra vilken (giltig) Python-kod som helst;
    • överföring, som transporterar data frĂ„n en plats till en annan, sĂ€g, MsSqlToHiveTransfer;
    • sensor kommer att tillĂ„ta att reagera eller försena fortsatt utförande av dag tills nĂ„got sker. HttpSensor kan slĂ„ an en specificerad endpoint, och nĂ€r den fĂ„r det svar den vĂ€ntar pĂ„, startar överföringen GoogleCloudStorageToS3Operator. En nyfiken sjĂ€l undrar: "Varför? Det gĂ„r ju att göra omstarter direkt i operatören!" Men för att inte blockera uppgiftspoolen med hĂ€ngande operatörer. Sensorn startar, kontrollerar och dör tills nĂ€sta försök.
  • Task — deklarerade operatörer oavsett typ som Ă€r kopplade till dags höjs till uppgiftsnivĂ„.
  • Uppgiftsinstans — nĂ€r generalplaneraren bestĂ€mmer sig för att det Ă€r dags att skicka uppgifter i strid till utförande-arbetare (direkt pĂ„ plats om vi anvĂ€nder LocalExecutor eller pĂ„ en fjĂ€rrnod i fallet med CeleryExecutor), tilldelar han dem en kontext (dvs ett set av variabler — utförandeparametrar), rullar ut kommandon eller förfrĂ„gningar och lĂ€gger dem i poolen.

Genererar uppgifter

LÄt oss först beskriva den övergripande schemat av vÄr dag, och sedan dyka djupare och djupare in i detaljerna, eftersom vi tillÀmpar vissa icke-triviala lösningar.

SÄ, i enklaste form kommer en sÄdan dag att se ut sÄ hÀr:

frÄn datetime import timedelta, datetime

frÄn airflow import DAG
frÄn airflow.operators.python_operator import PythonOperator

frÄn 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)

för conn_id, schema i sql_server_ds:
    PythonOperator(
        task_id=schema,
        python_callable=workflow,
        provide_context=True,
        dag=dag)

LÄt oss försöka förstÄ:

  • Först importerar vi de nödvĂ€ndiga biblioteken och nĂ„got mer;
  • sql_server_ds — detta Ă€r List[namedtuple[str, str]] med namnen pĂ„ anslutningarna frĂ„n Airflow Connections och databaserna som vi kommer att hĂ€mta vĂ„r tabell ifrĂ„n;
  • dag Ă€r deklarationen av vĂ„r DAG, som absolut mĂ„ste vara i globals(), annars hittar inte Airflow den. Vi mĂ„ste ocksĂ„ tala om för DAG:en:
    • vad den heter orders — detta namn kommer att dyka upp i webbgrĂ€nssnittet,
    • att den kommer att börja arbeta frĂ„n midnatt den 8 juli,
    • och den ska köras var sjĂ€tte timme (för de tuffa killarna hĂ€r istĂ€llet för timedelta() Ă€r det tillĂ„tet cron-strĂ€ng 0 0 0/6 ? * * *, för de mindre tuffa — ett uttryck som @daily);
  • workflow() gör det huvudsakliga arbetet, men inte nu. Just nu kommer vi bara att dumpa vĂ„r kontext i loggen.
  • Och nu lite enkel magi för att skapa uppgifter:
    • vi gĂ„r igenom vĂ„ra kĂ€llor;
    • initialiserar PythonOperator, som kommer att utföra vĂ„r tomma funktion workflow(). Glöm inte att ange ett unikt (inom DAG:en) uppgiftsnamn och koppla det till DAG:en. Flaggan provide_context kommer i sin tur att lĂ€gga extra argument i funktionen, som vi noga kommer att samla in med hjĂ€lp av **context.

Det Àr allt för nu. Vad har vi fÄtt:

  • en ny DAG i webbgrĂ€nssnittet,
  • ett par hundra uppgifter som kommer att köras parallellt (om Airflows, Celerys och servrarnas instĂ€llningar tillĂ„ter det).

NÄvÀl, vi har nÀstan fÄtt det.

Apache Airflow: gör ETL enklare
Vem ska sÀtta upp beroendena?

För att förenkla allt detta har jag infört docker-compose.yml bearbetningen requirements.txt pÄ alla noder.

Nu kör vi:

Apache Airflow: gör ETL enklare

GrĂ„ rutor — task instances, bearbetade av schemalĂ€ggaren.

Vi vÀntar lite, uppgifterna tas om hand av arbetarna:

Apache Airflow: gör ETL enklare

Gröna, förstĂ„s, — framgĂ„ngsrikt avklarade. Röda — mindre framgĂ„ngsrika.

Förresten, pĂ„ vĂ„r produktionsserver finns det ingen mapp .\/dags, som synkroniseras mellan maskinerna — alla DAG:ar ligger i git vĂ„r Gitlab, och Gitlab CI distribuerar uppdateringar till maskinerna vid sammanslagning i master.

Lite om Flower

Medan arbetarna sĂ€tter igĂ„ng vĂ„ra tomma uppgifter, lĂ„t oss pĂ„minna oss om ett annat verktyg, som kan visa oss nĂ„got — Flower.

Den allra första sidan med sammanfattande information om nod-arbetarna:

Apache Airflow: gör ETL enklare

Den mest omfattande sidan med uppgifter som skickats för bearbetning:

Apache Airflow: gör ETL enklare

Den mest trÄkiga sidan om statusen för vÄr mÀklare:

Apache Airflow: gör ETL enklare

Den mest fĂ€rgstarka sidan — med diagram över statusen för uppgifter och deras utförandetid:

Apache Airflow: gör ETL enklare

Laddar ner det som inte laddats ner

SĂ„, alla uppgifter har bearbetats, vi kan ta hand om de skadade.

Apache Airflow: gör ETL enklare

Och det visade sig vara en hel del skadade — av olika anledningar. Vid korrekt anvĂ€ndning av Airflow talar dessa rutor om att data verkligen inte har kommit fram.

Vi behöver kolla loggen och starta om de kraschade task instances.

Genom att klicka pÄ vilken ruta som helst ser vi de ÄtgÀrder som Àr tillgÀngliga för oss:

Apache Airflow: gör ETL enklare

Vi kan ta och göra Clear pÄ den som har kraschat. Det betyder att vi glömmer att nÄgot har gÄtt fel, och den samma task instansen gÄr tillbaka till schemalÀggaren.

Apache Airflow: gör ETL enklare

Det Ă€r uppenbart att göra sĂ„ med musen för alla röda rutor inte Ă€r sĂ€rskilt humant — det Ă€r inte det vi förvĂ€ntar oss av Airflow. Naturligtvis har vi ett massförstörelsevapen: Browse/Task Instances

Apache Airflow: gör ETL enklare

Vi vÀljer allt pÄ en gÄng och nollstÀller genom att trycka pÄ rÀtt alternativ:

Apache Airflow: gör ETL enklare

Efter rensning ser vÄra taxi ut sÄ hÀr (de vÀntar otÄligt pÄ att schemalÀggaren ska planera dem):

Apache Airflow: gör ETL enklare

Anslutningar, hooks och andra variabler

Det Àr dags att titta pÄ nÀsta 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("""Dear gentlemen, the reports have been updated"""),
    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("""
         Natasha, wake up, we have {{ dag.dag_id }} crashed
        """),
    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]

Har alla nÄgon gÄng gjort en rapportuppdatering? HÀr har vi den igen: det finns en lista över kÀllor för att hÀmta data; det finns en lista över vart den ska lÀggas; glöm inte att skicka ett meddelande nÀr allt har hÀnt eller gÄtt fel (men det handlar inte om oss, eller hur).

LÄt oss Äter gÄ igenom filen och titta pÄ de nya oklara sakerna:

  • from commons.operators import TelegramBotSendMessage — ingenting hindrar oss frĂ„n att skapa vĂ„ra egna operatörer, vilket vi har nyttjat genom att skapa ett litet wrapper för att skicka meddelanden i Unblocked. (Om denna operatör kommer vi att prata mer om nedan);
  • default_args={} — DAG:en kan dela samma argument med alla sina operatörer;
  • to='{{ var.value.all_the_kings_men }}' — fĂ€ltet till kommer inte att vara hĂ„rdkodad, utan kommer att formas dynamiskt med hjĂ€lp av Jinja och en variabel med en lista av e-postadresser som jag noggrant lagt i Admin/Variables;
  • trigger_rule=TriggerRule.ALL_SUCCESS — villkor för operatörens start. I vĂ„rt fall kommer brevet skickas till cheferna endast om alla beroenden har arbetat framgĂ„ngsrikt;
  • tg_bot_conn_id='tg_main' — argumenten conn_id tar emot identifierare för anslutningar som vi skapar i Admin/Connections;
  • trigger_rule=TriggerRule.ONE_FAILED — meddelanden till Telegram kommer att skickas endast vid förekomst av misslyckade uppgifter;
  • task_concurrency=1 — vi förhindrar samtidigt körning av flera uppgiftsinstanser av samma uppgift. Annars fĂ„r vi flera VerticaOperator (som tittar pĂ„ en tabell);
  • report_update >> [email, tg] — alla VerticaOperator kommer att konvergera i att skicka brevet och meddelandet, sĂ„ hĂ€r:
    Apache Airflow: gör ETL enklare

    Men eftersom notifieringsoperatörerna har olika startvillkor, kommer endast en att fungera. I Tree View ser det nÄgot mindre överskÄdligt ut:
    Apache Airflow: gör ETL enklare

Jag ska sĂ€ga ett par ord om makron och deras vĂ€nner — av variabler.

Makron Àr Jinja-platsinnehÄllare som kan infoga olika anvÀndbar information i operatörsargument. Till exempel sÄ hÀr:

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

{{ ds }} kommer att expandera till innehÄllet i kontextvariabeln execution_date i formatet YYYY-MM-DD: 2020-07-14. Det mest trevliga Àr att kontextvariabler fÀsts vid en viss instans av en uppgift (en ruta i Tree View), och vid omstart kommer platsinnehÄllarna att visa samma vÀrden.

De tilldelade vÀrdena kan ses med hjÀlp av Rendered-knappen pÄ varje uppgiftsinstans. SÄ hÀr ser det ut för uppgiften med att skicka brevet:

Apache Airflow: gör ETL enklare

Och sÄ hÀr ser det ut för uppgiften med att skicka meddelandet:

Apache Airflow: gör ETL enklare

En fullstÀndig lista över inbyggda makron för den senaste tillgÀngliga versionen finns hÀr: Macros Reference

Dessutom kan vi genom plugins deklarera egna makron, men det Àr en helt annan historia.

Förutom de fördefinierade objekten kan vi anvÀnda vÀrden frÄn vÄra variabler (jag har redan gjort detta i koden ovan). LÄt oss skapa i Admin/Variables ett par objekt:

Apache Airflow: gör ETL enklare

Klart, nu kan vi anvÀnda det:

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

I vÀrdet kan det vara en skalar eller sÄ kan det ligga en JSON. I fallet med JSON:

bot_config

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

anvÀnder vi bara sökvÀgen till nyckeln vi behöver: {{ var.json.bot_config.bot.token }}.

Jag ska sÀga ett enda ord och visa en skÀrmdump om kopplingar. HÀr Àr allt elementÀrt: pÄ sidan Admin/Connections skapar vi en koppling, lÀgger in vÄra anvÀndarnamn/lösenord och mer specifika parametrar. SÄ hÀr:

Apache Airflow: gör ETL enklare

Lösenord kan krypteras (mer noggrant Ă€n i standardversionen), och man kan Ă€ven vĂ€lja att inte ange typ av koppling (som jag gjorde för tg_main) — poĂ€ngen Ă€r att listan över typer Ă€r inbĂ€ddad i Airflows modeller och kan inte Ă€ndras utan att gĂ„ in i kĂ€llkoden (om jag missat nĂ„got i min forskning, ber jag er rĂ€tta mig), men att fĂ„ kredentialer enbart genom namn Ă€r inget som hindrar oss.

Och man kan ocksÄ skapa flera kopplingar med samma namn: i sÄ fall kommer metoden BaseHook.get_connection(), som hÀmtar kopplingar efter namn, att returnera en slumpmÀssig av flera likadana (det skulle vara mer logiskt att göra Round Robin, men vi lÀmnar det till Airflows utvecklare).

Variables och Connections Ă€r utan tvekan anvĂ€ndbara verktyg, men det Ă€r viktigt att inte förlora balansen: vilka delar av dina flöden förvarar du faktiskt i koden och vilka ger du till Airflow. Å ena sidan kan det vara bekvĂ€mt att snabbt Ă€ndra vĂ€rden, till exempel e-postadress, via UI. Å andra sidan Ă€r det trots allt att Ă„tergĂ„ till att klicka med musen, vilket vi (jag) ville bli av med.

Arbete med kopplingar Àr en av uppgifterna hooks. Generellt sett Àr Airflow-hakar kopplingspunkter till externa tjÀnster och bibliotek. Till exempel kommer JiraHook att öppna en klient för att interagera med Jira (vi kan flytta uppgifter fram och tillbaka), och med hjÀlp av SambaHook kan vi ladda upp en lokal fil till smb-punkten.

Analyserar en anpassad operator

Och vi har kommit till den punkt dÀr vi kan titta pÄ hur TelegramBotSendMessage

Kod commons/operators.py Àr uppbyggd med sjÀlva operatören:

frÄn typning importera Union

frÄn airflow.operators importera BaseOperator

frÄn commons.hooks importera TelegramBotHook, TelegramBot

klass TelegramBotSendMessage(BaseOperator):
    """Skicka meddelande till chat_id med TelegramBotHook

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

    def __init__(self,
                 chat_id: Union[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'Send "{self.message}" till chatten {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

HÀr, precis som allt annat i Airflow, Àr allt vÀldigt enkelt:

  • Vi Ă€rver frĂ„n BaseOperator, som implementerar ganska mĂ„nga Airflow-specifika saker (kolla pĂ„ din fritid)
  • Vi deklarerade fĂ€lten template_fields, dĂ€r Jinja kommer att leta efter makron för bearbetning.
  • Vi organiserade korrekta argument för __init__(), och satte standarder dĂ€r det behövdes.
  • Vi glömde inte att initiera överklagandet.
  • Vi öppnade den relevanta hooken TelegramBotHook, fick klientobjektet frĂ„n den.
  • Vi Ă„sidosatte (överskred) metoden BaseOperator.execute(), som Airflow kommer att kalla pĂ„ nĂ€r det Ă€r dags att köra operatorn — hĂ€r genomför vi den huvudsakliga Ă„tgĂ€rden, utan att glömma att logga in. (Förresten, vi loggar direkt i stdout och stderr — Airflow fĂ„ngar allt, snyggt kapslar in det och sorterar det var det behövs.)

LÄt oss se vad vi har i commons/hooks.py. Första delen av filen, med sjÀlva hooken:

frÄn typning importera Union

frÄn airflow.hooks.base_hook importera BaseHook
frÄn requests_toolbelt.sessions importera BaseUrlSession

klass TelegramBotHook(BaseHook):
    """Telegram Bot API hook

    Notera: lÀgg till en anslutning med tom Conn Typ och glöm inte
    fylla i 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

Jag vet egentligen inte vad jag kan förklara hÀr, jag noterar bara viktiga punkter:

  • Vi Ă€rver, tĂ€nker pĂ„ argumenten — i de flesta fall kommer det att vara ett enda: conn_id;
  • Vi Ă„sidosĂ€tter standardmetoder: jag begrĂ€nsade mig till get_conn(), dĂ€r jag fĂ„r anslutningsparametrarna efter namn och bara plockar ut sektionen extra (detta fĂ€lt Ă€r för JSON), dĂ€r jag (enligt min egen instruktion!) lade Telegram-bot tokenet: {"bot_token": "YOuRAwEsomeBOtToKen"}.
  • Jag skapar en instans av vĂ„r TelegramBot, och ger den redan en specifik token.

Det var allt. Du kan fÄ klienten frÄn hooken med hjÀlp av TelegramBotHook().clent eller TelegramBotHook().get_conn().

Och den andra delen av filen, dÀr jag gör ett mikrounpakning för Telegram REST API, sÄ att jag inte behöver dra med mig samma python-telegram-bot för bara en metod sendMessage.

class TelegramBot:
    """Telegram Bot API wrapper

    Exempel:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hej, Àlskling')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hej, Àlskling', 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))

Den rĂ€tta vĂ€gen Ă€r att sammanfoga allt detta: TelegramBotSendMessage, TelegramBotHook, TelegramBot — i ett plugin, lĂ€gga det i ett offentligt repository, och ge ut det som Open Source.

Medan vi studerade allt detta har vÄra rapportuppdateringar lyckats krasch och skickat ett felmeddelande till mig i kanalen. Jag ska gÄ och kolla vad som Àr fel igen...

Apache Airflow: gör ETL enklare
I vÄrt dag har nÄgot gÄtt sönder! Var det inte precis det vi vÀntade pÄ? Precis!

Ska du hÀlla?

KÀnner du att jag missade nÄgot? Jag lovade ju att föra data frÄn SQL Server till Vertica, och nu tog jag och gled ifrÄn Àmnet, skamligt!

Detta var ett medvetet brott, jag var bara tvungen att förklara lite terminologi för er. Nu kan vi fortsÀtta.

VÄr plan var sÄ hÀr:

  1. Skapa ett dag
  2. Generera uppgifter
  3. Se hur allt ser vackert ut
  4. Tilldela sessionnummer till uppladdningar
  5. HÀmta data frÄn SQL Server
  6. LĂ€gga data i Vertica
  7. Samla statistik

SÄ, för att köra allt detta, gjorde jag en liten tillÀgg till vÄr 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

HĂ€r installerar vi:

  • Vertica som vĂ€rd dwh med de mest standardinstĂ€llda konfigurationerna,
  • tre instanser av SQL Server,
  • fyller databaserna med lite data (titta absolut inte i mssql_init.py!)

Vi startar allt med hjÀlp av en nÄgot mer komplex kommando Àn förra gÄngen:

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

Vad vÄr fantastiska randomizer har genererat kan vi, med hjÀlp av alternativet Data Profiling/Ad Hoc Query:

Apache Airflow: gör ETL enklare
Det viktigaste Àr att inte visa detta för analytikerna

Jag kommer inte att gÄ in pÄ detaljer om ETL-sessionerna dÀr, allt Àr trivialt: vi skapar en databas, i den en tabell, vi omsluter allt i en kontextmanager, och nu gör vi sÄ hÀr:

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

    # Lasta arbetsflödet
    ...

    session.successful = True
    session.loaded_rows = 15

session.py

frÄn sys import stderr

class Session:
    """ETL arbetsflödes session

    Exempel:
        med Session(task_name) som session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Bra jobbat'
    """

    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 = """
            SKAPA TABELL OM INTE FINNS sessions (
                id          SERIAL       INTE NULL PRIMARY KEY,
                task_name   VARCHAR(200) INTE NULL,

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

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

    def open(self):
        query = """
            INFÖR sessions (task_name, finished)
            VÄRDEN (%s, NULL)
            RETURNERAR id;
            """
        self._id = self._execute(query, self.task_name)
        print(self, 'öppnad')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Session Àr inte öppen')
        query = """
            UPPDATERA sessions
            STÄLL IN
                finished    = DEFAULT,
                successful  = %s,
                loaded_rows = %s,
                comment     = %s
            VAR
                id = %s
            RETURNERAR id;
            """
        self._execute(query, self.successful, self.loaded_rows,
                      self.comment, self.id)
        print(self, 'stÀngd',
              ', framgÄngsrik: ', self.successful,
              ', Laddad: ', self.loaded_rows,
              ', kommentar:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

Det Àr dags att hÀmta vÄr data frÄn vÄra mer Àn hundra och femtio tabeller. Vi gör detta med hjÀlp av nÄgra mycket enkla rader:

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

query = f"""
    VÄLJ 
        id, start_time, end_time, type, data
    FRÅN dbo.Orders
    DÄR
        CONVERT(DATE, start_time) = '{dt}'
    """

df = pd.read_sql_query(query, source_conn)
  1. Med hjÀlp av hooken hÀmtar vi frÄn Airflow pymssql-anslutning
  2. I frĂ„gan lĂ€gger vi in en begrĂ€nsning i form av ett datum — i funktionen kommer templatern att fĂ„nga det.
  3. Vi matar in vĂ„r frĂ„ga pandas, som kommer att hĂ€mta för oss DataFrame — den kommer att vara till nytta för oss senare.

Jag anvÀnder ersÀttning {dt} istÀllet för frÄgeparametern %s inte för att jag Àr en elak Buratino, utan för att pandas kan inte hantera pymssql och ger den sista en urchin params: List, Àven om den verkligen vill tuple.
Notera ocksÄ att utvecklaren pymssql beslutat att inte stödja den lÀngre, och det Àr dags att flytta till pyodbc.

LÄt oss se vad Airflow har fyllt argumenten för vÄra funktioner med:

Apache Airflow: gör ETL enklare

Om det inte finns nÄgra data, finns det ingen mening att fortsÀtta. Men att betrakta uppladdningen som framgÄngsrik Àr ocksÄ konstigt. Men det Àr inte ett fel. Vad ska man göra?! SÄ hÀr:

if df.empty:
    raise AirflowSkipException('Inga rader att ladda')

AirflowSkipException kommer att sÀga att det faktiskt inte finns nÄgot fel, och vi hoppar över uppgiften. I grÀnssnittet kommer det att vara varken en grön eller röd ruta, utan en rosa.

LÄt oss ge vÄra data nÄgra kolumner:

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

Det vill sÀga:

  • DB:n vi hĂ€mtade bestĂ€llningarna frĂ„n,
  • Identifieraren för vĂ„r uppladdande session (den kommer att vara olika för varje uppgift),
  • Hash frĂ„n kĂ€llan och bestĂ€llningsidentifieraren — sĂ„ att vi har en unik identifierare för bestĂ€llningen i den slutgiltiga databasen (dĂ€r allt kommer att samlas i en tabell).

KvarstĂ„r Ă€r nĂ€st sista steget: ladda allt i Vertica. Och, som det Ă€r konstigt, Ă€r en av de mest effektiva sĂ€tten att göra detta — via 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)
  1. Vi gör en speciell mottagare StringIO.
  2. pandas som vÀnligt lÀgger vÄr DataFrame i form av CSV-rader.
  3. Vi öppnar en anslutning till vÄr Àlskade Vertica med ett hook.
  4. Och nu med hjÀlp av copy() ska vi skicka vÄra data direkt till Vertica!

FrÄn drivrutinen hÀmtar vi hur mÄnga rader som har laddats, och vi sÀger till sessionschefen att allt Àr OK:

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

Det var allt.

I produktionen skapar vi mÄltabellen manuellt. HÀr har jag tillÄtit mig en liten automatisering:

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)

Jag anvÀnder VerticaOperator() för att skapa databasschemat och tabellen (om de inte redan finns, förstÄs). Det viktigaste Àr att stÀlla in beroenden korrekt:

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

Sammanfattning

— NĂ„vĂ€l, — sa musungen, — Ă€r det inte sant att nu
Har du försÀkrat dig om att jag Àr det lÀskigaste djuret i skogen?

Julia Donaldson, "Gruffalon"

Jag tror att om jag och mina kollegor tĂ€vlade om vem som snabbast kunde skapa och köra en ETL-process frĂ„n grunden: de med sina SSIS och musen och jag med Airflow
 Och sedan skulle vi jĂ€mföra hur lĂ€tt det var att underhĂ„lla
 Åh, jag tror att ni skulle hĂ„lla med om att jag skulle vinna pĂ„ alla fronter!

Om vi ska vara lite mer seriösa, sĂ„ har Apache Airflow – genom att beskriva processerna i form av programkod – gjort mitt jobb mycket mer bekvĂ€mt och trevligare.

Dess oĂ€ndliga skalbarhet: bĂ„de vad gĂ€ller plugins och dess benĂ€genhet för skalbarhet – ger dig möjlighet att anvĂ€nda Airflow praktiskt taget överallt: oavsett om det Ă€r i hela cykeln för insamling, förberedelse och behandling av data, eller i raketuppskjutningar (till Mars, förstĂ„s).

Avslutande del, referens och information

Fallgropar som vi har samlat för er

  • start_date. Ja, det Ă€r redan ett lokalt meme. Genom huvudargumentet i DAG:en start_date gĂ„r allt igenom. Kort sagt, om man anger i start_date det aktuella datumet, och i schedule_interval – en dag, sĂ„ kommer DAG:en att köras imorgon tidigast.
    start_date = datetime(2020, 7, 7, 0, 1, 2)

    Och inga fler problem.

    Det finns ocksÄ ett annat exekveringsfel kopplat till detta: Task is missing the start_date parameter, vilket oftast betyder att du har glömt att knyta tillverkaren till DAG:en.

  • Allt pĂ„ en maskin. Ja, bĂ„de databasen (bĂ„de Airflow och vĂ„rt grĂ€nssnitt), webbservern, schemalĂ€ggaren och arbetarna. Det fungerade till och med. Men med tiden vĂ€xte antalet uppgifter för tjĂ€nsterna, och nĂ€r PostgreSQL började ge svar baserat pĂ„ index efter 20 ms istĂ€llet för 5 ms, tog vi bort den.
  • LocalExecutor. Ja, vi Ă€r fortfarande kvar pĂ„ den, och vi har redan kommit till avgrunden. LocalExecutor:n har varit tillrĂ€cklig för oss, men nu Ă€r det dags att expandera med minst en arbetare, och vi mĂ„ste anstrĂ€nga oss för att flytta till CeleryExecutor. Och eftersom man kan arbeta med den pĂ„ en enda maskin, sĂ„ finns det inget som hindrar anvĂ€ndningen av Celery Ă€ven pĂ„ en server som "naturligtvis aldrig kommer att gĂ„ i produktion, jag lovar!"
  • Icke-anvĂ€ndning av inbyggda verktyg:
    • Connections för att lagra tjĂ€nsternas autentiseringsuppgifter,
    • SLA Misses för att reagera pĂ„ uppgifter som inte har körts i tid,
    • XCom för att utbyta metadat (jag sa metadata!) mellan uppgifterna i DAG:en.
  • Missbruk av e-post. Vad kan man sĂ€ga? Jag hade stĂ€llt in aviseringar för alla upprepningar av fallna uppgifter. Nu har jag över 90k e-postmeddelanden frĂ„n Airflow i min arbets-Gmail, och webbgrĂ€nssnittet för e-post vĂ€grar att ta bort mer Ă€n 100 Ă„t gĂ„ngen.

Fler fallgropar: Apache Airflow Pitfalls

Verktyg för Ànnu mer automatisering

För att vi ska kunna arbeta mer med huvudet istÀllet för med hÀnderna har Airflow förberett följande Ät oss:

  • REST API — den har fortfarande status som Experimental, vilket inte hindrar den frĂ„n att fungera. Med dess hjĂ€lp kan man inte bara fĂ„ information om DAG:ar och uppgifter, utan ocksĂ„ stoppa/starta en DAG, skapa en DAG Run eller pool.
  • CLI — mĂ„nga verktyg Ă€r tillgĂ€ngliga via kommandoraden som inte bara Ă€r obekvĂ€ma att anvĂ€nda via WebUI utan faktiskt inte finns dĂ€r. Till exempel:
    • backfill Ă€r nödvĂ€ndigt för att köra instanser av uppgifter pĂ„ nytt.
      Till exempel kommer analytiker och sĂ€ger: "Men ni, kĂ€ra vĂ€nner, har skrĂ€p i data frĂ„n den 1 till 13 januari! ÅtgĂ€rda det!" Och du Ă€r sĂ„dĂ€r, hokus pokus:
      airflow backfill -s '2020-01-01' -e '2020-01-13' orders
    • DatabasunderhĂ„ll: initdb, resetdb, upgradedb, checkdb.
    • run, som lĂ„ter dig starta en instans av en uppgift och skita i alla beroenden. Dessutom gĂ„r det att starta den genom LocalExecutor, Ă€ven om du har ett Celery-kluster.
    • NĂ€stan samma sak gör test, men den skriver ingenting i databasen.
    • connections möjliggör massbelĂ€ggning av anslutningar via shell.
  • Python API — en ganska hardcore metod för interaktion som Ă€r avsedd för plugins, inte att pilla med för hand. Men vem hindrar oss frĂ„n att gĂ„ in i /home/airflow/dags, starta ipython och börja göra som vi vill? Man kan till exempel exportera alla anslutningar med följande kod:
    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)
  • Anslutning till Airflows metadatabas. Jag rekommenderar inte att skriva till den, men det gĂ„r betydligt snabbare och enklare att hĂ€mta uppgiftstillstĂ„nd för olika specifika metrik Ă€n genom nĂ„got av API:erna.

    LĂ„t oss sĂ€ga att inte alla vĂ„ra uppgifter Ă€r idempotenta och ibland kan falla, och det Ă€r okej. Men flera fel — det Ă€r redan misstĂ€nkt och man borde kolla upp det.

    Akta dig för SQL!

    MED last_executions SOM (
SELECT
    task_id,
    dag_id,
    execution_date,
    state,
        row_number()
        OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC) AS rn
FROM public.task_instance
WHERE
    execution_date > now() - INTERVAL '2' DAGAR
),
failed SOM (
    SELECT
        task_id,
        dag_id,
        execution_date,
        state,
        CASE WHEN rn = row_number() OVER (
            PARTITION BY task_id, dag_id
            ORDER BY execution_date DESC)
                 THEN TRUE END AS last_fail_seq
    FROM last_executions
    WHERE
        state IN ('failed', 'up_for_retry')
)
SELECT
    task_id,
    dag_id,
    count(last_fail_seq)                       AS unsuccessful,
    count(CASE WHEN last_fail_seq
        AND state = 'failed' THEN 1 END)       AS failed,
    count(CASE WHEN last_fail_seq
        AND state = 'up_for_retry' THEN 1 END) AS up_for_retry
FROM failed
GROUP BY
    task_id,
    dag_id
HAVING
    count(last_fail_seq) > 0

LĂ€nkar

SjÀlvklart, de första tio lÀnkarna frÄn Googles sökresultat innehÄller mappen Airflow frÄn mina bokmÀrken.

Och lÀnkarna som anvÀnds i artikeln:

KĂ€lla: habr.com

Köp pĂ„litlig webbhotell med DDoS-skydd, VPS VDS-servrar đŸ”„ Köp pĂ„litlig webbhotell med DDoS-skydd, VPS VDS-servrar | ProHoster