Apache Airflow: lihtsustame ETL-i

Tere, mina olen Dmitri Logvinenko — andmete insener ettevõtte «Vezёт» analüüsiosakonnas.

Räägin teile suurepärasest tööriistast ETL-protsesside arendamiseks — Apache Airflow. Kuid Airflow on nii universaalne ja mitmekesine, et tasub sellele tähelepanu pöörata isegi siis, kui te ei tegele andmevoogudega, vaid vajate aeg-ajalt mõnede protsesside käivitamist ja nende teostamise jälgimist.

Ja jah, ma mitte ainult ei räägi, vaid ka näitan: programmis on palju koodi, ekraanipilte ja soovitusi.

Mida tavaliselt näed, kui googled sõna Airflow / Wikimedia Commons

Sisukord

• Sissejuhatus
• Peamine, praktiline (ja natuke teoreetiline) osa
  • Miks see meile (ja teile) vajalik on
  • Kogume klastrit
  • Põhimõisted
  • Genereerime ülesandeid
  • Natuke Flowerist
  • Laadime alla alalaadimise
  • Ühendused, hookid ja muud muutujad
  • Täitame kohandatud operaatorit
  • Kas sa teed seda?
  • Teeme kokkuvõtte
• Viimane, teaduslik-informatiivne osa
  • Kohad, mille me teie jaoks kokku korjasime
  • Veel suurema automatiseerimise vahendid
• Viidatud lingid

Sissejuhatus

Apache Airflow — ta on nagu Django:

• kirjutatud Pythonis,
• on suurepärane adminpaneel,
• piiramatu laiendatavus,

— ainult parem, ja loodud hoopis teiste eesmärkide saavutamiseks, nimelt (kui on kirjutatud enne katti):

• ülesannete käivitamine ja jälgimine piiramatul arvul masinatel (nii palju kui teie lubab Celery/Kubernetes ja teie südametunnistus)
• dünaamilise töövoo genereerimine väga lihtsalt kirjutatava ja tajutava Python-koodi kaudu
• ja võimalus siduda omavahel ükskõik millised andmebaasid ja API-d nii valmis komponentide kui ka isetehtud pluginatega (mida on äärmiselt lihtne teha).

Me kasutame Apache Airflow'i nii:

• kogume andmeid erinevatest allikatest (palju SQL Serveri ja PostgreSQL instantsse, erinevaid API-sid rakenduste mõõdikute jaoks, isegi 1C) DWH-sse ja ODS-i (meie puhul on need Vertica ja Clickhouse).
• nagu arenenud cron, mis käivitab andmete konsolideerimise protsesse ODS-is ja jälgib nende hooldust.

Kuni hiljutise ajani katab meie vajadusi üks väike server 32 tuuma ja 50 GB RAM-iga. Airflow'is töötab sel juhul:

• üle 200 DAG-i (nende töövood, kuhu oleme ülesandeid täitnud),
• igaühes keskmiselt 70 ülesannet,
• selle headuse käivitamine (ka keskmiselt) korra tunni jooksul.

Ja sellest, kuidas me laienesime, kirjutan ma allpool, kuid praegu määratlegem über-ülesanne, mida me lahendama hakkame:

On each of the three source SQL Servers, there are 50 databases — instances of one project, so their structure is identical (almost everywhere, muah-ha-ha), which means each has a table called Orders (after all, a table with that name can fit into any business). We pull data, adding metadata fields (source server, source database, ETL task identifier) and naively drop them into, say, Vertica.

Lähme!

Peamine, praktiline (ja natuke teoreetiline) osa

Miks see meile (ja teile) vajalik on

When the trees were tall, and I was just a simple SQL-worker in a Russian retail company, we rolled out ETL processes aka data streams using the two tools available to us:

• Informatica Power Center — an extremely sprawling system, highly productive, with its own hardware, its own versioning. I used maybe 1% of its capabilities. Why? Well, first, this interface is from the noughties and mentally weighed down on us. Secondly, this thing is designed for incredibly complex processes, fierce component reuse, and other very-important-enterprise-features. Let's not even mention that it costs as much as the wing of an Airbus A380/year.

  Caution, the screenshot may hurt people under 30 a bit.

  

• SQL Server Integration Server — we used this fellow in our internal project streams. Well, in fact: we already use SQL Server, and not using its ETL tools would be unreasonable. Everything about it is good: the interface is beautiful, and the execution reports… But that’s not why we love software products, oh no. We can version it dtsx (which is an XML with nodes mixed when saving); but what's the use? And to create a task package that will move a hundred tables from one server to another? You’d drop your index finger on the mouse button before moving twenty. But it definitely looks more stylish:

  

We were definitely looking for ways out. It even almost came to a self-made SSIS package generator…

… and then I found a new job. And on that job, I encountered Apache Airflow.

When I learned that describing ETL processes is just simple Python code, I almost danced with joy. This is how data streams underwent versioning and diffusion, and pouring tables with a uniform structure from a hundred databases into one target became a task for Python code on one and a half to two 13" screens.

Kogume klastrit

Ärge korraldage lastaeda ja rääkige täiesti ilmsetest asjadest, nagu Airflow'i, teie valitud andmebaasi, Celery ja muude asjade seadistamisest, millest dokumentatsioonis räägitakse.

Kuna me saame kohe katsetama hakata, olen ma koostanud docker-compose.yml kuna:

• Käivitame tegelikult Airflow: Scheduler, Webserver. Seal käivitub ka Flower Celery ülesannete jälgimiseks (sest see on juba lükatud apache/airflow:1.10.10-python3.7, ja me ei ole selle vastu);
• PostgreSQL, kuhu Airflow kirjutab oma haldusandmed (planeerija andmed, täitmise statistika jne), ja Celery tähistab lõpetatud ülesandeid;
• Redis, mis toimib Celery ülesannete vahendajana;
• Celery worker, mis hakkab ülesandeid tegelikult täitma.
• Kausta . /dags salvestame meie DAGide kirjeldusfailid. Need tuuakse automaatselt üles, seega pole vaja iga kord kogu steki täiesti üles tõsta.

Kohati on kood näidetes osaliselt toodud (et mitte teksti ummistada) ja mõned kohad muudetakse protsessi käigus. Täielikke töötavaid koodinäiteid saab vaadata hoidlas. https://github.com/dm-logv/airflow-tutorial.

docker-compose.yml

versioon: '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:
      <<: *airflow-config

      AIRFLOW__SCHEDULER__DAG_DIR_LIST_INTERVAL: 30
      AIRFLOW__SCHEDULER__CATCHUP_BY_DEFAULT: 'False'
      AIRFLOW__SCHEDULER__MAX_THREADS: 8

      AIRFLOW__WEBSERVER__LOG_FETCH_TIMEOUT_SEC: 10

    depends_on:
      - airflow-db
      - broker

    command: >
      -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:
      <<: *airflow-config

    command: >
      -c " sleep 10 &&
           pip install --user -r \/requirements.txt &&
           \/entrypoint worker"

    depends_on:
      - airflow
      - airflow-db
      - broker

Märkused:

• Ma kompositsiooni koostamisel toitusin peamiselt tuntud pildist puckel\/docker-airflow – kindlasti vaadake. Võib-olla ei vaja te elus midagi muud.
• Kõik Airflow seaded on saadaval mitte ainult läbi airflow.cfg, vaid ka keskkonnamuutujate kaudu (tänu arendajatele), mida ma häbiväärselt kasutasin.
• Muidugi ei ole see production ready: ma ei seadnud konteineritele heartbeate, ma ei vaevanud end turvalisusega. Kuid miinimum, mis sobib meie katsetele, on loodud.
• Pange tähele, et:
  • DAGide kaust peab olema kergesti ligipääsetav nii ajakava koostajale kui ka töötajatele.
  • Sama kehtib kõigi kolmandate osapoolte teekide kohta — need peavad kõik olema installitud ajakava koostaja ja töötajate masinatesse.

Nüüd aga lihtsalt:

$ docker-compose up --scale worker=3

Kui kõik on käima lükatud, saab veebiliideseid vaatama minna:

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

Põhimõisted

Kui te ei saanud nendest "dagidest" midagi aru, siis siin on lühike sõnastik:

• Scheduler — peamine tegelane Airflow's, kes kontrollib, et robotid töötaksid, mitte inimene: jälgib ajakava, uuendab dag'e, käivitab ülesandeid.

  Tegelikult olid vanade versioonide puhul tal mäluprobleemid (ei, mitte amneesia, vaid lekked) ja konfiguratsioonides jäi isegi alles üks legassi parameeter run_duration — tema taaskäivitamise intervall. Aga nüüd on kõik korras.

• DAG (ehk "dag") — "suunatud ahelate graaf", kuid see määratlemine ei ütle kellelegi suurt midagi, aga sisuliselt on see konteiner omavahel suhtlevatele ülesannetele (vt allpool) või analoog Paketile SSIS-is ja Töökargile Informatica-s.

  Lisaks dag'idele võivad olla ka subdag'id, aga tõenäoliselt me needeni ei jõua.

• DAG Run — initsialiseeritud dag, millele on antud oma execution_date. Ühe dag'i dag'runnid võivad täiesti töötada paralleelselt (kui olete muidugi teinud oma ülesanded idempotentseteks).
• Operator — see on koodi tükk, mis vastutab mingi konkreetse toimingu täitmise eest. On kolm tüüpi operaatorit:
  • tegevus, nagu näiteks meie lemmik PythonOperator, mis on võimeline täitma mis tahes (kehtivat) Python-koodi;
  • ülekandmine, mis viivad andmeid ühelt poolt teisele, näiteks MsSqlToHiveTransfer;
  • sensor lubab reageerida või peatada edasise dag'i täitmise, kuni mingi sündmus ei ole toimunud. HttpSensor võib küsida määratud lõpp-punkti, ja kui ootab vajaliku vastuse, käivitab körvülekande GoogleCloudStorageToS3Operator. Uudishimu küsib: "miks? Kas poleks lihtsam teha kordusi otse operaatoris!" Ja siis, et mitte koormata ülesannete kogumit seisatavate operaatoritega. Sensor käivitub, kontrollib ja sureb kuni järgmise katseni.
• Task — väljakuulutatud operaatorid, olenemata tüübist, ja dag'iga seotud ülesannete määramine tõstetakse ülesande tasemele.
• Task instance — kui peaplaneerija otsustas, et ülesanded on valmis lähetamiseks tote jooksutavatele töötajatele (otse kohapeal, kui kasutame LocalExecutor või kaug-noodi puhul CeleryExecutor), määrab ta neile konteksti (st muutuja-sätte komplekti), rakendab käskude või päringute malle ja kogub need kokku.

Genereerime ülesandeid

Esmalt määratleme meie dag'i üldise skeemi, seejärel süveneme järjest enam detailidesse, sest rakendame mitmeid mitte triviaalsetes lahendustes.

Nii et kõige lihtsamas vormis näeb selline dag välja nii:

import timedelta, datetime

from airflow import DAG
from airflow.operators.python_operator import PythonOperator

from 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)

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

Hakkame siis üheskoos välja selgitama:

• Esmalt impordime vajalikud teegid ja veel mõned muud asjad;
• sql_server_ds — see on List[namedtuple[str, str]] Airflow Connections'i ühenduste nimed ja andmebaasid, millest me meie tabelit toome;
• dag on meie DAG'i deklaratsioon, mis peab olema globals(), muidu Airflow seda ei leia. DAG'ile tuleb ka öelda:
  • kuidas seda kutsuda orders see nimi ilmub hiljem veebiliideses,
  • et see hakkab töötama alates 8. juuli keskööst,
  • ja käivitama ta peaks umbes iga 6 tunni järel (karmimatele on siin 'timedelta()' asemel lubatud -string, kergematele - väljend nagu cron@daily 0 0 0/6 ? * * *workflow() teeb peamise töö, aga mitte praegu. Praegu viskame lihtsalt meie konteksti logisse.);
• Nüüd lihtne maagia ülesannete loomisel: jalutame meie allikate kaudu;
• initsialiseerime
  • , mis täidab meie anomaalia,
  • . Ärge unustage määrata ülesande ainulaadne (DAG'i ulatuses) nimi ja seondada see DAG'iga. Lipp PythonOperatorprovide_context Nüüd lihtne maagia ülesannete loomisel:toob omakorda funktsiooni lisaargumente, mille me hoolikalt kogume, kasutades **context Sellega on praegu kõik. Mida me saime: uus DAG veebiliideses,.

poolteist sada ülesannet, mis töötavad paralleelselt (kui Airflow, Celery ja serverite seadistused seda lubavad).

• Noh, peaaegu saime.
• Kes seab sõltuvused paika?

Selle asja lihtsustamiseks lülitasin selle

käsitlemiseks

kõikidesse sõlmedesse. docker-compose.yml Nüüd siis läks lahti: requirements.txt Hallid ruudud - ülesande instantsid, mida planeerija on töödelnud.

Ootame natuke, ülesanded haaravad töötajad:

Rohelised, nagu mõistetav, - edukalt lõpule viidud. Punased - mitte nii edukalt.

Muide, meie tootel ei ole mingit kausta,

, mis süsynkroniseerib masinate vahel - kõik DAG'id asuvad

meie Gitlab'is, ja Gitlab CI paigutab uuendused masinatesse, kui toimub liitmine . /dagsKui töötajad töötlevad meie tühiseid ülesandeid, tuletame meelde veel ühte tööriista, mis meile midagi näidata võib - Flower. git Esimene leht, kus on kokkuvõtvad andmed sõlmedest - töötajatest: master.

Natuke Flowerist

Kõige jaotatum leht ülesannetest, mis on tööle saadetud:

Esimene leht, mis sisaldab kokkuvõtlikku teavet töötajate sõlmpunktide kohta:

Rikkaim leht ülesannetega, mis on saadetud töötlemiseks:

Kõige igavam leht meie maakleri seisundi kohta:

Kõige värvilisem leht — ülesannete seisundi graafikute ja nende täitmise ajaga:

Laadime alla alalaadimise

Nii et kõik ülesanded on töötanud, saame haavatud ära viia.

Haavatuid oli üllatavalt palju — erinevatel põhjustel. Kui Airflowd kasutatakse õigesti, siis need ruudud näitavad, et andmed ei ole kindlasti kohale jõudnud.

Peame vaatama logisid ja taaskäivitama nurjunud ülesande näidiseid.

Klikkides igal ruudul, näeme meie käsutuses olevaid toiminguid:

Saame võtta ja teha nurjunule Clear. See tähendab, et unustame, et seal on mingi probleem ja sama ülesande näidis suundub planeerijale.

Selge, et ei ole väga inimsõbralik kõiki punaseid ruute hiirega nii käsitleda — seda me Airflowlt ei oota. Loomulikult on meil massihävitusrelv: Browse/Task Instances

Valime kõik korraga ja nullime, valides õige punkti:

Pärast puhastamist näevad meie taksod välja nii (nad ootavad juba, et planeerija nad ajastaks):

Ühendused, hookid ja muud muutujad

On paras aeg vaadata järgmist DAG-i, 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("""Lugupeetud, raportid on uuendatud"""),
    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, ärka üles, meil on {{ dag.dag_id }} kukkunud
        """),
    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]

Kas kõik on kunagi teinud raportite uuendamist? See on jälle see: on nimekiri allikatest, kust andmeid võtta; on nimekiri, kuhu need panna; ärge unustage märku anda, kui kõik juhtus või läks rikki (noh, see ei puuduta meid, ei).

Vaatame taas faili ja uurime uusi arusaamatuid asju:

• from commons.operators import TelegramBotSendMessage — ei ole midagi, mis takistaks meil oma operaatorite loomist, mida me tegime, luues väikese mähise sõnumite saatmiseks Unblockitud. (Sellest operaatorist räägime ka hiljem);
• default_args={} — DAG võib jagada samu argumendi kõigile oma operaatoritele;
• to='{{ var.value.all_the_kings_men }}' — väli to meil ei ole seda kõvasti määratletud, vaid see genereeritakse dünaamiliselt Jinja ja e-mailide nimekirja muutuja abil, mille ma hoolikalt asetasin Admin/Variables;
• trigger_rule=TriggerRule.ALL_SUCCESS — operaatori käivitamise tingimus. Meie puhul saadetakse kiri juhtidele ainult siis, kui kõik sõltuvused töötasid edukalt;
• tg_bot_conn_id='tg_main' — argumendid conn_id võtavad endasse ühenduste identifikaatorid, mille me loome Admin/Connections;
• trigger_rule=TriggerRule.ONE_FAILED — Telegrami sõnumid saadetakse ainult, kui on katkestatud ülesandeid;
• task_concurrency=1 — keelame sama ülesande mitme ülesande eksemplari samaaegse käivitamise. Vastasel juhul saame mitu korraga VerticaOperator (mis vaatavad ühte tabelit);
• report_update >> [email, tg] — kõik VerticaOperator kohtuvad kirja ja sõnumi saatmisel, just nii:
  

  Kuid kuna teavitajate operaatoritel on erinevad käivitamistingimused, töötab ainult üks. Puustruktuuris näeb see välja vähem selge:
  

Räägin paar sõna makrode ja nende sõprade — muutujad.

Makrod on Jinja-plekistrid, mis võivad esitada erinevat kasulikku teavet operaatorite argumentidesse. Näiteks nii:

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

{{ ds }} lahti muudetakse konteksti muutuja sisuks execution_date Port 2222 YYYY-MM-DD: 2020-07-14. Kõige parem on see, et konteksti muutujaid seotakse kindlalt konkreetse ülesande eksemplariga (ruudukesega Puustruktuuris), ja uuesti käivitamisel avanevad plekistrid samadele väärtustele.

Määratud väärtusi saab vaadata nuppuga Rendered igal ülesande eksemplaril. Näiteks on see ülesandel, mis saadab kirja:

Ja see ülesandel, mis saadab sõnumi:

Täielik nimekiri sisseehitatud makrodest viimase saadaval oleva versiooni jaoks on siin: Macros Reference

Veelgi enam, pluginade abil saame kuulutada oma makrosid, kuid see on juba hoopis teine lugu.

Lisaks eelnevalt määratud asjadele saame kasutada oma muutujaid (muutsin neid ülaltoodud koodis). Loome Admin/Variables paar asja:

Küll, nüüd saab hakata kasutama:

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

Mõisted võivad olla skalaarsed või JSON. JSON-i puhul:

bot_config

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

kasutame lihtsalt vajalikku võtme teed: {{ var.json.bot_config.bot.token }}.

Kõnestan vaid ühe sõna ja näitan ühte ekraanipilti seoses ühendustega. Siin on kõik elementaarne: lehe peal Admin/Connections loome ühenduse, paneme sinna oma kasutajanimed/paroolid ja veel spetsiifilisemad seaded. Nii:

Paroolid saab krüpteerida (rohkem kui vaikimisi juhul) või pole määratud ühenduse tüüpi (nagu tegin tg_main) — asi on selles, et tüübid on Airflow mudelites sisse ehitatud ja ilma lähtekoodidesse sekkumata neid ei saa muuta (kui midagi on vale, siis palun parandage mind), kuid krüpteerida nime järgi pole meil midagi takistust.

Ja lisaks võime luua mitu sama nimega ühendust: sel juhul meetod BaseHook.get_connection(), mis toob meile ühendused nime järgi, annab juhusliku ühe mitme sarnase seast (oleks loogilisem teha Round Robin, kuid jätame selle Airflow arendajate mureks).

Muutujad ja ühendused on kahtlemata suurepärased tööriistad, kuid oluline on mitte kaotada tasakaalu: millised teie voogude osad säilitad konkreetselt koodis ning millised usaldad Airflow’d. Ühelt poolt on mugav kiirelt muuta väärtust, näiteks postkasti, läbi UI. Teiselt poolt on see ikka naasmine hiireklikkide juurde, millest tahtsime (mina) lahti saada.

Ühendustega töötamine on üks ülesanne hookide paralleelseks täitmiseks.. Üldiselt on Airflow hook'id ühenduspunktid kolmandate teenuste ja raamatukogudega. Näiteks JiraHook avab meile kliendi suhtlemiseks Jira'ga (saame ülesandeid liigutada edasi-tagasi), ja SambaHook võimaldab me pushida kohaliku faili smb-punkti.

Täitame kohandatud operaatorit

Ja oleme lähenenud sellele, et vaadata, kuidas on tehtud TelegramBotSendMessage

Kood commons/operators.py oma operaatori kohta:

importida Union

from airflow.operators import BaseOperator

from commons.hooks import TelegramBotHook, TelegramBot

class TelegramBotSendMessage(BaseOperator):
    """Saada sõnum chat_id-le kasutades TelegramBotHook'i

    Näide:
        >>> 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 }} ebaõnnestus :(',
        ...     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'Saadan "{self.message}" chati {self.chat_id}')
        self.client.send_message(chat_id=self.chat_id,
                                 message=self.message)

Siin, nagu kõik Airflow's, on kõik väga lihtne:

• Oleme pärinud BaseOperator, mis rakendab palju Airflow-spetsiifilisi asju (vaadake kunagi, kui on aega)
• Oleme kuulutanud välja väljad template_fields, kus Jinja otsib makrosid töötlemiseks.
• Oleme korraldanud õiged argumendid __init__(), seadnud vaikimisi väärtused sinna, kus on vajalik.
• Ka eelneva initsialiseerimist ei unustanud.
• Oleme avanud vastava hook'i TelegramBotHook, saanud sellelt kliendi objekti.
• Olemegi üle kirjutanud (override) meetodi BaseOperator.execute(), mida Airflow kutsub välja, kui on aeg operaatorit tööle panna — just seal me teeme põhitegevuse, unustamata logida. (Me logime, muide, otse stdout ja stderr — Airflow teeb kõik püüdmiseks ja pakib ilusasti kokku, paneb õigesse kohta.)

Vaatame, mis meil on commons/hooks.py. Faili esimene osa, koos hook'iga:

importida Union

from airflow.hooks.base_hook import BaseHook
from requests_toolbelt.sessions import BaseUrlSession

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

    Märkus: lisage ühendus tühja Conn Type'iga ja ärge unustage
    täita 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

Ma ei tea isegi, mida siit selgitada, lihtsalt märgin olulised punktid:

• Pärime, mõtleme argumentidele — enamasti on neid üks: conn_id;
• Üle kirjutame standardsed meetodid: ma piirdusin get_conn(), kus ma saan ühenduse parameetrid nime järgi ja lihtsalt tõstan sektsiooni. extra (see code for JSON), where I put the Telegram bot token according to my own instructions: {"bot_token": "YOuRAwEsomeBOtToKen"}.
• Creating an instance of our TelegramBot, giving it a specific token.

That's it. You can get the client from the hook using TelegramBotHook().client või TelegramBotHook().get_conn().

And the second part of the file, where I wrap the Telegram REST API to avoid carrying the same python-telegram-bot for just one method sendMessage.

class TelegramBot:
    """Telegram Bot API wrapper

    Examples:
        >>> TelegramBot('YOuRAwEsomeBOtToKen', '@myprettydebugchat').send_message('Hi, darling')
        >>> TelegramBot('YOuRAwEsomeBOtToKen').send_message('Hi, darling', 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))

The correct approach is to put all of this in: TelegramBotSendMessage, TelegramBotHook, TelegramBot — a plugin, store it in a public repository, and release it as Open Source.

While we were studying all this, our report updates successfully piled up and sent me a message about an error in the channel. I will go check what went wrong again...

Something broke in our DAG! Wasn't this what we were waiting for? Exactly!

Kas sa teed seda?

Do you feel like I've missed something? I promised to transfer data from SQL Server to Vertica, and here I went off-topic, how careless!

This misdeed was intentional; I simply had to explain some terminology to you. Now we can move on.

Our plan was as follows:

1. Create a DAG
2. Generate tasks
3. See how everything looks nice
4. Assign session numbers to uploads
5. Fetch data from SQL Server
6. Store data in Vertica
7. Compile statistics

So, to run all this, I made a small addition to our 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

Seal tõstame:

• Vertica kui host dwh kõige vaikimisi seadistustega,
• kolm SQL Serveri eksemplari,
• täiendame andmebaase viimasel ajal mõne andmega (ärge mingil juhul vaadake sisse mssql_init.py!)

Käivitame kogu selle hea kraami veidi keerukama käsuga kui eelmisel korral:

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

Mis meie imetore genereerija tootis, on võimalik, kasutades punkti Data Profiling/Ad Hoc Query:

Peamine, et seda analüütikutele ei näidata

Detailid ETL-seanssides ma ei peatuks, seal on kõik triviaalne: loome andmebaasi, sinna tabeli, katame kõik konteksti juhiga, ja nüüd teeme nii: with Session(task_name) as session: print('Load', session.id, 'started')# Load workflow ...session.successful = True session.loaded_rows = 15

session.py

session.py

from sys import stderr

class Session:
    """ETL töövoo seanss

    Näide:
        with Session(task_name) as session:
            print(session.id)
            session.successful = True
            session.loaded_rows = 15
            session.comment = 'Hea töö'
    """

    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, 'avatud')
        return self

    def close(self):
        if not self._id:
            raise SessionClosedError('Seanss ei ole avatud')
        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, 'suletud',
              ', edukas: ', self.successful,
              ', Laaditud: ', self.loaded_rows,
              ', kommentaar:', self.comment)

class SessionError(Exception):
    pass

class SessionClosedError(SessionError):
    pass

Kätte on jõudnud aeg andmed kokku koguda meie poolteise saja tabeli hulgast. Teeme seda väga lihtsalt ridadega:

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)

1. Hüüdme abil saame Airflow'st pymssql-ühenduse
2. Küsitlusse lisame kuupäeva piirangu - selle edastab meile mallitegija.
3. Käivitame meie päringu pandas, mis tõmbab meile DataFrame — see tuleb meile hiljem kasuks.

Kasutame asendust {dt} küsi parameetri asemel %s mitte sellepärast, et ma olen kuri Buratino, vaid pigem sellepärast, et pandas ei suuda hakkama saada pymssql ja paneb viimasele params: Loend, kuigi see väga tahab tulp.
Pange tähele, et arendaja pymssql otsustas teda enam mitte toetada, ja on aeg kolida pyodbc.

Vaatame, millega Airflow täitis meie funktsioonide argumendid:

Kui andmeid ei olnud, siis pole mõtet jätkata. Aga ka selle üle lugemine, et laadimine õnnestus, on kummaline. Kuid see pole viga. A-a-a, mida teha?! Siin on, mida:

if df.empty:
    raise AirflowSkipException('Ridasid ei ole laadimiseks')

AirflowSkipException ütleb Airflow, et viga ei ole, ja ülesanne jääb vahele. Liideses ei ole roheline ja ei punane ruut, vaid roosa.

Lisame meie andmetele mõned veerud:

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

Täpselt:

• DB, kust me tellimused võtsime,
• Meie laadimisessiooni identifikaator (see on erinev iga ülesande jaoks),
• Allika ja tellimuse identifikaatori hash — et lõpp-andmebaasis (kus kõik valatakse ühte tabelisse) oleks meil ainulaadne tellimuse identifikaator.

Jäänud on eelviimane samm: laadida kõik Vertica'sse. Ja, nagu ei oleks uskumatu, on üks efektiivseid viise teha seda — läbi 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. Teeme spetsiaalse vastuvõtja StringIO.
2. pandas mis lahendab meie DataFrame välise lisandina CSV-rid.
3. Avame ühenduse meie lemmik Vertica hook'iga.
4. Ja nüüd saame copy() saata meie andmed otse Vertica'sse!

Draiverist võtame, kui palju ridu laaditi, ja ütleme sessiooni juhile, et kõik on OK:

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

See on kõik.

Produksioonis loome sihttableti käsitsi. Siin lubasin endale väikese automaatika:

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)

Ma kasutan VerticaOperator() loomiseks andmebaasi skeemi ja tabeli (kui need veel puuduvad, loomulikult). Peamine on õigesti seada sõltuvused:

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

Teeme kokkuvõtte

— No nii, — ütles hiirepoeg, — ei ole tõsi, et nüüd
Kas sa oled veendunud, et metsa kõige hirmsam loom olen mina?

Julia Donaldson, „Gruffalo“

Mõtle, kui me kolleegidega korraldaksime konkursi: kes suudab kiiremini luua ja käivitada ETL-protsessi nullist: nemad oma SSIS-i ja hiirega ja mina Airflow'iga... Ja siis võrdleksime veel hooldamise mugavust... Oh, ma arvan, et oled nõus, et ma möödun neist igas osas!

Kui rääkida natuke tõsisemalt, siis Apache Airflow – tänu protsesside kirjeldamisele programmikoodina – tegi mu töö märksa mugavamaks ja meeldivamaks.

Selle piiramatud laienemisvõimalused: nii pluginate osas kui ka skaleerimisvõimes — annavad võimaluse rakendada Airflow’d praktiliselt igas valdkonnas: olgu see siis andmete kogumise, ettevalmistamise ja töötlemise täissükkel või isegi rakettide käivitamine (loomulikult Marsile).

Viimane, teaduslik-informatiivne osa

Kohad, mille me teie jaoks kokku korjasime

• start_date. Jah, see on juba kohalik meem. Peamise DAG-argumenti kaudu start_date läbivad kõik. Lühidalt: kui määrata start_date praegune kuupäev ja schedule_interval — üks päev, siis DAG käivitub homme mitte varem.

  start_date = datetime(2020, 7, 7, 0, 1, 2)

  Ja rohkem pole mingeid probleeme.

  Sellega on seotud veel üks täitmisviga: Task is missing the start_date parameter, mis enamikul juhtudest ütleb, et unustasid DAG-i operaatoriga siduda.

• Kõik ühel masinal. Jah, ja andmebaasid (nii Airflow enda kui ka meie ümbruse), ja veebiserver, ja ajastaja, ja töötajad. Ja see isegi toimis. Aga aja jooksul kasvas teenuste ülesannete arv, ja kui PostgreSQL hakkas vastama indeksilt 20 ms asemel 5 ms, võtsime selle ja viisime minema.
• LocalExecutor. Jah, me kasutame seda endiselt ja oleme juba ääre peal. LocalExecutori jaoks on meil siiani piisavalt, kuid nüüd on aeg vähemalt ühe töötaja võrra laiendada, ja tuleb pingutada, et üle minna CeleryExecutori peale. Ja arvestades, et sellega saab töötada ka ühel masinal, ei takista miski Celery kasutamist isegi mitte serveris, mis "loomulikult ei tule kunagi tootmisse, tõotame!"
• Kasutamata sisekasutuse:
  • Connections teenuste autentimisandmete haldamiseks,
  • SLA Misses ülesannete jaoks, mis ei töötanud õigeaegselt,
  • XCom metatekstide vahetamiseks (ma ütlesin metateavet!) DAG-i ülesannete vahel.
• Posti kuritarvitamine. Mis siin ikka öelda? Olin seadnud teavitused kõikide korduvate ebaõnnestunud ülesannete jaoks. Nüüd on minu töö Gmailis üle 90 000 e-kirja Airflow'lt ning postiteenuse veebiväljaanne keeldub kustutamast rohkem kui 100 korraga.

Rohkem peidetud kivisid: Apache Airflowi Probleemid

Veel suurema automatiseerimise vahendid

Selleks, et saaksime veel rohkem mõelda, mitte ainult kätega töötada, on Airflow meie jaoks ette valmistanud järgmised võimalused:

• REST API — tal on endiselt Eksperimentaalse staatuse, mis ei takista selle toimimist. Selle abil on võimalik mitte ainult saada teavet DAG-ide ja ülesannete kohta, vaid ka peatada/taasalustada DAG-i, luua DAG Run või baas.
• cli-runtime'is ja kubectl'is — käsurealt on saadaval palju tööriistu, mis ei ole lihtsalt ebamugavad veebiliidese kaudu kasutada, vaid pole seal isegi saadaval. Näiteks:
  • backfill vajalik, et käivitada ülesannete instantside uuesti käivitamine.
    Näiteks tulid analüütikud ja ütlesid: „Teie, seltsimees, andmetes on jama 1. kuni 13. jaanuarini! Parandage!“. Ja siis sa ütled:

    airflow backfill -s '2020-01-01' -e '2020-01-13' orders

  • Andmebaasi hooldus: initdb, resetdb, upgradedb, checkdb.
  • run, mis võimaldab käivitada ühe ülesande instantsi, jättes kõik sõltuvused kõrvale. Veelgi enam, seda saab käivitada läbi LocalExecutor, isegi kui sul on Celery klastri.
  • Umbes sama asja teeb test, ainult et see ei kirjuta andmebaasi.
  • connections võimaldab massiliselt luua ühendusi shellist.
• Python API on üsna keeruline suhtlemisviis, mis on mõeldud pistikprogrammidena, mitte selle käsitsi mudimiseks. Aga kes meid takistab minemast /home/airflow/dags, käivitama ipython ja hakkama siin mängima? Näiteks võib kõiki ühendusi eksportida järgmise koodiga:

  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)

• Ühendus Airflow'i metadatuuri andmebaasi. Ma ei soovita sinna kirjutada, kuid ülesannete olekute saamine erinevate spetsiifiliste metrikate jaoks on oluliselt kiirem ja lihtsam kui läbi ühegi API.

  Ütleme nii, et kaugeltki kõik meie ülesanded ei ole idempotentsed ning võivad mõnikord ebaõnnestuda, ja see on normaalne. Aga mitu ebaõnnestumist on juba kahtlane ja tuleks kontrollida.

  Ole ettevaatlik, SQL!

  VIIMASTE_EXECUTIONIDEGA AS (
  VALI
      ülesande_id,
      dag_id,
      täitmise_aeg,
      olek,
          rida_numbrina()
          ÜLE (
              JAOTUSEKS ülesande_id, dag_id
              KORRALDA täitmise_aeg LANGUS) NII rn
  FROM public.task_instance
  KUS
      täitmise_aeg > nüüd() - INTERVALL '2' PÄEVA
  ),
  ebaõnnestunud AS (
      VALI
          ülesande_id,
          dag_id,
          täitmise_aeg,
          olek,
          JUHTUM KUI rn = rida_numbrina() ÜLE (
              JAOTUSEKS ülesande_id, dag_id
              KORRALDA täitmise_aeg LANGUS)
                   SIIS TÕENE LÕPP JN kui ebaõnnestunud
      KUS
          olek IN ('ebaõnnestunud', 'ootab_kordamist')
  )
  VALI
      ülesande_id,
      dag_id,
      loe(ebaõnnestunud) AS ebaõnnestunud,
      loe(JUHTUM KUI ebaõnnestunud
          JA olek = 'ebaõnnestunud' SIIS 1 LÕPP) AS ebaõnnestunud,
      loe(JUHTUM KUI ebaõnnestunud
          JA olek = 'ootab_kordamist' SIIS 1 LÕPP) AS ootab_kordamist
  FROM ebaõnnestunud
  GRUPEERI
      ülesande_id,
      dag_id
  OLLES
      loe(ebaõnnestunud) > 0

Viidatud lingid

Ja loomulikult esimesed kümme linki Google'i otsingust, mis viivad minu Airflow kaustadesse.

• Apache Airflow Dokumentatsioon — loomulikult peaks alustama ametlikust dokumentatsioonist, aga kes neid juhiseid üldiselt loeb?
• Parimad tavad — no vähemalt loe loo autoreid nende soovitusi.
• Airflow kasutajaliides — algus: kasutajaliides piltides
• Apache Airflow'i põhikontseptide mõistmine — hästi kirjeldatud põhikontseptsioonid, kui sa (juhtumisi!) ei saanud aru, mis ma ütlen.
• Tianlong'i blogi — Kuidas ehitada Airflow serverit/klaastri — lühike juhend Airflow klastrite seadistamiseks.
• Apache Airflow jooksutamine Lyftis — peaaegu sama huvitav artikkel, kuid formaalsust on rohkem ja näiteid vähem.
• Kuidas Apache Airflow jaotab ülesandeid Celery töötajatele — töös koos Celeryga.
• DAG-i kirjutamise parimad tavad Apache Airflow's — idempotentsete ülesannete, ID kaudu laadimise ning muude huvitavate asjade kohta.
• Sõltuvuste haldamine Apache Airflow's — ülesannete sõltuvused ja Trigger Rule, mida mainisin ainult möödaminnes.
• Airflow: Kui sinu DAG on kaugel graafikust — kuidas ületada mõningaid 'töötab nagu kavandatud' plaanija puhul, laadida kaotatud andmeid ja seada ülesannetele prioriteete.
• Kasulikud SQL-päringud Apache Airflow'le — kasulikud SQL-päringud Airflow metainfo kohta.
• Alustage töövoogude arendamist Apache Airflow's — seal on kasulik jaotis kohandatud sensori loomise kohta.
• Fetchr andeteaduse infrastruktuuri loomine AWS-is Presto ja Airflow'ga — huvitav lühike märkus andeteaduse infrastruktuuri ülesehitamisest AWS-is.
• 7 levinud viga, mida uurides Airflow DAG-e kontrollida — levinud vead (kui keegi ikka ei loe juhiseid).
• Salvestage ja pääsete ligi paroolile, kasutades Apache Airflow'i — naeratage, kuidas inimesed paroolide hoidmise juures jalge alla saadavad, kuigi saate lihtsalt kasutada Ühendusi.
• Pythoni ja Apache Airflow'i zeni — varjatud DAG-i edastus, konteksti edastamine funktsioonis, taas sõltuvustest, ja ka ülesannete käivitamise vahelejätmisest.
• Airflow: Vähem Tuntud Näpunäited, Trikid ja Parimad Tavad — kasutamise kohta vaikimisi argumentide ja params malle ning ka muutujaid ja ühendusi.
• Airflow Scheduleri profiilimine — jutt sellest, kuidas plaani koostaja Airflow 2.0 jaoks ette valmistatakse.
• Apache Airflow koos 3 Celery töölauaga docker-compose'is — veidi vananenud artikkel meie klastri kasutusele võtmisest version: '1' services: simplesample-sonar: image: sonarqube:lts ports: - 9001:9000 - 9092:9092 network_mode: bridge.
• 4 Mallimise Ülesannet, Kasutades Airflow Konteksti — dünaamilised ülesanded mallide ja konteksti edastamise abil.
• Veateated Airflow'is — standardsed ja kohandatud teated e-posti ja Slacki kaudu.
• Airflow Workshop: keerulised DAG-id ilma kottide kasutamiseta — Ülesannete lõhestamine, makrod ja XCom.

Ja lingid, mis on artiklis kasutatud:

• Makrode viidatud — saadaval mallides kasutamiseks kohandatud kohad.
• Tavalised Lood — Airflow — Levinud vead DAGide loomisel.
• puckel/docker-airflow: Docker Apache Airflow — version: '1' services: simplesample-sonar: image: sonarqube:lts ports: - 9001:9000 - 9092:9092 network_mode: bridge eksperimentideks, silumiseks ja muuks.
• python-telegram-bot/python-telegram-bot: Oleme loonud sulle paketi, millele ei saa vastu seista — Pythoni pakett Telegrami REST API jaoks.

Allikas: habr.com