🥇Airflow — alat za udobno i brzo razvijanje i održavanje batch-procesa obrade podataka

Bok, Habr! U ovom članku želim govoriti o jednom izvanrednom alatu za razvoj batch procesa obrade podataka, na primjer, u infrastrukturi korporativnog DWH-a ili vašem DataLake-u. Riječ je o Apache Airflow-u (dalje Airflow). Nepravedno je zapostavljen na Habu, i u glavnom dijelu ću pokušati uvjeriti vas da barem obratite pažnju na Airflow pri odabiru planera za vaše ETL/ELT procese.

Ranije sam pisao seriju članaka na temu DWH-a kada sam radio u Tinkoff banci. Sada sam dio tima Mail.Ru grupe i bavim se razvojem platforme za analizu podataka u industriji igara. Kako se bude pojavljavale vijesti i zanimljiva rješenja, moj tim i ja ćemo ovdje govoriti o našoj platformi za analizu podataka.

Prolog

Dakle, hajde da počnemo. Što je Airflow? To je biblioteka (ili skup biblioteka) za razvoj, planiranje i praćenje radnih procesa. Glavna značajka Airflow-a: za opisivanje (razvoj) procesa koristi se kod na Python jeziku. Iz toga proistječe mnoštvo prednosti za organizaciju vašeg projekta i razvoj: u suštini, vaš (na primjer) ETL projekt je samo Python projekt, i možete ga organizirati kako vam odgovara, uzimajući u obzir osobitosti infrastrukture, veličinu tima i druge zahtjeve. Instrumentalno je sve jednostavno. Koristite, na primjer, PyCharm + Git. To je divno i vrlo praktično!

Sada razmotrimo glavne entitete Airflow-a. Razumijevanje njihove suštine i namjene optimizirat će arhitekturu procesa. Vjerojatno je glavni entitet - to je usmjereni aciklični graf (dalje DAG).

DAG

DAG je određena smisleno povezivanje vaših zadataka, koje želite izvršiti u strogo određenoj sekvenci prema određenom rasporedu. Airflow nudi prikladan web sučelje za rad s DAG-ovima i drugim entitetima:

DAG može izgledati ovako:

Programer, projektirajući DAG, ugrađuje skup operatora na kojima će se graditi zadaci unutar DAG-a. Ovdje dolazimo do još jednog važnog entiteta: Airflow Operator.

Operatori

Operator je entitet na temelju kojeg se kreiraju instance zadatka, gdje se opisuje što će se događati tijekom izvršenja instance zadatka. Izdanja Airflow-a s GitHub-a već sadrže skup operatora spremnih za korištenje. Primjeri:

BashOperator — operator za izvođenje bash naredbi.
PythonOperator — operator za pozivanje Python koda.
EmailOperator — operator za slanje emaila.
HTTPOperator — operator za rad s http zahtjevima.
SqlOperator — operator za izvršavanje SQL koda.
Sensor — operator koji čeka na događaj (naredni trenutak, pojavu potrebne datoteke, reda u bazi podataka, odgovor iz API-ja — itd.).

Postoje specifičniji operatori: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

Također možete razvijati operatore prema svojim potrebama i koristiti ih u projektu. Na primjer, stvorili smo MongoDBToHiveViaHdfsTransfer, operator za izvoz dokumenata iz MongoDB u Hive, i nekoliko operatora za rad s ClickHouse: CHLoadFromHiveOperator i CHTableLoaderOperator. U suštini, kada se u projektu pojavi često korišteni kod temeljen na osnovnim operatorima, možete razmisliti o sastavljanju u novi operator. To će olakšati daljnji razvoj i obogatiti vašu biblioteku operatora u projektu.

Nakon toga, svi ti primjeri zadataka trebaju se izvršiti, a sada ćemo razgovarati o planeru.

Planer

Planer zadataka u Airflowu temelji se na Celery. Celery je Python biblioteka koja omogućava organizaciju reda i asinkrono i distribuirano izvršavanje zadataka. S aspekta Airflowa, svi zadaci se dijele na poolove. Poolove se obično kreiraju ručno. Njihova svrha je ograničiti opterećenje na radu sa izvorom ili klasificirati zadatke unutar DWH-a. Poolovima se može upravljati putem web sučelja:

Svaki pool ima ograničenje broja slotova. Prilikom stvaranja DAG-a dodjeljuje mu se pool:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

Pool postavljen na razini DAG-a može se prepisati na razini zadatka.
Za planiranje svih zadataka u Airflowu odgovoran je poseban proces — Scheduler. U suštini, Scheduler se bavi svojom mehanikom postavljanja zadataka na izvršavanje. Zadak, prije nego što uđe u izvršavanje, prolazi kroz nekoliko faza:

U DAG-u su izvršeni prethodni zadaci, novi se može staviti u red.
Red se sortira u zavisnosti od prioriteta zadatka (prioritetima se takođe može upravljati), i, ako u grupi postoji slobodan slot, zadatak se može preuzeti na rad.
Ako postoji slobodan Celery worker, zadatak se usmerava na njega; počinje rad koji ste programirali u zadatku, koristeći određeni operator.

Relativno je jednostavno.

Scheduler radi na više svih DAG-ova i svih zadataka unutar DAG-ova.

Da bi Scheduler započeo rad s DAG-om, DAG-u je potrebno odrediti raspored:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

Postoji set gotovih presetova: @once, @hourly, @daily, @weekly, @monthly, @yearly.

Takođe se mogu koristiti cron-izrazi:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*\/10 * * * *')

Datum izvršenja

Da biste shvatili kako funkcioniše Airflow, važno je razumeti šta je datum izvršenja za DAG. U Airflow-u, DAG ima dimenziju datuma izvršenja, tj. u zavisnosti od rasporeda rada DAG-a, kreiraju se primerci zadataka za svaku datu izvršenja. I za svaku datu izvršenja zadaci se mogu ponovo izvršiti — ili, na primer, DAG može raditi istovremeno u više datuma izvršenja. Ovo je jasno prikazano ovde:

Nažalost (a možda i srećom: zavisi od situacije), ako se menja implementacija zadatka u DAG-u, izvršavanje u prethodnim datumima izvršenja će se već vršiti uzimajući u obzir ispravke. Ovo je dobro ako je potrebno ponovo izračunati podatke u prethodnim periodima novim algoritmom, ali loše, jer se gubi ponovljivost rezultata (naravno, niko vas ne sprečava da vratite potrebnu verziju iz Git-a i jednom izvršite ono što je potrebno, onako kako je potrebno).

Generisanje zadataka

Implementacija DAG-a je kod na Python-u, tako da imamo vrlo zgodan način da smanjimo obim koda prilikom rada, na primer, s shardiranim izvorima. Recimo da imate tri sharda MySQL kao izvor, i potrebna su vam podaci iz svakog. Pritom, nezavisno i paralelno. Kod na Python-u u DAG-u može izgledati ovako:

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='\/data\/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:]},
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

DAG se dobija ovakav:

U ovom procesu možete dodati ili ukloniti shard jednostavno podešavanjem postavki i ažuriranjem DAG-a. Praktično!

Možete koristiti i složeniju generaciju koda, na primjer rad s izvorima u obliku Baze podataka ili opisivanje tablične strukture, algoritma rada s tablicom i, uzimajući u obzir posebnosti infrastrukture DWH, generirati proces učitavanja N tablica u vaše skladište. Ili, na primjer, rad s API-jem koji ne podržava rad s parametrima u obliku liste; možete generirati N zadataka u DAG-u prema toj listi, ograničiti paralelizam zahtjeva u API-u s pulom i izvući potrebne podatke iz API-ja. Fleksibilno!

Repozitorij

Airflow ima svoj backend repozitorij, Bazu podataka (može biti MySQL ili Postgres, mi koristimo Postgres) u kojoj se pohranjuju stanja zadataka, DAG-ova, postavke veza, globalne varijable itd. Ovdje bismo htjeli reći da je repozitorij u Airflowu vrlo jednostavan (oko 20 tablica) i praktičan ako želite izgraditi neki svoj proces preko njega. Sjeća se 100500 tablica u repozitoriju Informatica, koje su trebale dugo vremena da se razumiju prije nego što se shvati kako izgraditi upit.

Monitoring

Uzimajući u obzir jednostavnost repozitorija, možete sami izgraditi proces praćenja zadataka koji vam odgovara. Koristimo bilježnicu u Zeppelin-u, gdje pratimo stanje zadataka:

To može biti i web sučelje samog Airflowa:

Airflow kod je otvoren, pa smo mi kod nas dodali aleriranje u Telegram. Svaki radni instanca zadatka, ako dođe do greške, šalje poruke u grupu na Telegramu u kojoj je cijeli razvojni i support tim.

Kroz Telegram dobivamo brzu reakciju (ako je potrebna), a putem Zeppelin-a opću sliku o zadacima u Airflow-u.

U konačnici

Airflow je prvenstveno open-source, i ne morate čekati čuda od njega. Budite spremni uložiti vrijeme i napor u izgradnju funkcionalnog rješenja. Cilj je dostižan, vjerujte mi, isplati se. Brzina razvoja, fleksibilnost, jednostavnost dodavanja novih procesa — sviđaće vam se. Naravno, potrebno je posvetiti više pažnje organizaciji projekta, stabilnosti rada samog Airflow-a: čuda ne postoje.

Trenutno naš Airflow svakodnevno obrađuje oko 6,5 tisuća zadataka. Po karakteru su prilično različiti. Ima zadataka učitavanja podataka u glavno DWH iz mnogih različitih i vrlo specifičnih izvora, ima zadataka izračuna vitrina unutar glavnog DWH, ima zadataka publikacije podataka u brzo DWH, ima mnogo, mnogo različitih zadataka — i Airflow sve to prožvače dan za danom. Ako govorimo brojkama, to je 2,3 tisuće ELT zadataka različite složenosti unutar DWH (Hadoop), oko 2,5 stotina baza podataka izvora, to je tim od četiri ETL programera, koji su podijeljeni na ETL procesiranje podataka u DWH i na ELT procesiranje podataka unutar DWH, a naravno i jednog administratora, koji se bavi infrastrukturom usluge.

Planovi za budućnost

Broj procesa neizbježnije raste, i glavno što ćemo raditi u vezi s infrastrukturom Airflowa — to je skaliranje. Želimo izgraditi Airflow klaster, odvojiti nekoliko čvorova za worker-e Celery i napraviti duplicirajuću instancu s procesima planiranja zadataka i repozitorijem.