Ciao, Habr! In questo articolo voglio parlarvi di uno strumento straordinario per lo sviluppo di processi batch di elaborazione dati, ad esempio, nell'infrastruttura di un DWH aziendale o del vostro DataLake. Parleremo di Apache Airflow (di seguito Airflow). È ingiustamente trascurato su Habr, e nella parte principale cercherò di convincervi che almeno Airflow merita di essere considerato nella scelta di un pianificatore per i vostri processi ETL/ELT.
In precedenza ho scritto una serie di articoli sul tema DWH quando lavoravo in Tinkoff Bank. Ora sono diventato parte del team di Mail.Ru Group e mi occupo dello sviluppo di una piattaforma per l'analisi dei dati nel settore dei giochi. Pertanto, man mano che emergono notizie e soluzioni interessanti, io e il mio team condivideremo qui la nostra piattaforma per l'analisi dei dati.
Prologo
Iniziamo. Cos'è Airflow? È una libreria (o un ) per la creazione, la pianificazione e il monitoraggio dei flussi di lavoro. La caratteristica principale di Airflow è che i processi sono definiti (sviluppati) utilizzando codice in Python. Questo comporta numerosi vantaggi per l'organizzazione del tuo progetto e dello sviluppo: in sostanza, il tuo progetto ETL, ad esempio, è semplicemente un progetto Python, e puoi organizzarlo come preferisci, tenendo conto delle caratteristiche infrastrutturali, delle dimensioni del team e di altri requisiti. Strumentalmente, è tutto semplice. Utilizza, ad esempio, PyCharm + Git. È fantastico e molto comodo!
Adesso esaminiamo le principali entità di Airflow. Comprendendo la loro essenza e il loro scopo, potrai organizzare al meglio l'architettura dei processi. Probabilmente, l'entità principale è il Directed Acyclic Graph (DAG).
DAG
Il DAG è un certo insieme concettuale delle attività che desideri eseguire in una sequenza rigorosamente definita secondo un programma prestabilito. Airflow offre un'interfaccia web comoda per lavorare con i DAG e altre entità:
Un DAG può apparire in questo modo:
Un sviluppatore, progettando un DAG, definisce un insieme di operatori su cui saranno costruiti i task all'interno del DAG. Qui arriviamo a un'altra entità importante: l'Operatore di Airflow.
Operatori
L'Operatore è un'entità su cui si basano le istanze dei task, descrivendo quali azioni verranno eseguite durante l'esecuzione dell'istanza del task. già contengono un insieme di operatori pronti all'uso. Esempi:
- BashOperator — operatore per eseguire comandi bash.
- PythonOperator — operatore per chiamare codice Python.
- EmailOperator — operatore per inviare email.
- HTTPOperator — operatore per gestire richieste http.
- SqlOperator — operatore per eseguire codice SQL.
- Sensor — operatore che attende un evento (l'arrivo del momento giusto, la comparsa di un file richiesto, una riga in un database, una risposta da un'API, ecc.).
Ci sono operatori più specifici: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.
Puoi anche sviluppare operatori personalizzati in base alle tue esigenze e utilizzarli nel progetto. Ad esempio, abbiamo creato MongoDBToHiveViaHdfsTransfer, un operatore per esportare documenti da MongoDB a Hive, e diversi operatori per lavorare con : CHLoadFromHiveOperator e CHTableLoaderOperator. In sostanza, non appena nel progetto emerge un codice frequentemente utilizzato, costruito su operatori di base, si può pensare di racchiuderlo in un nuovo operatore. Questo semplificherà lo sviluppo futuro e arricchirà la tua libreria di operatori nel progetto.
Successivamente, tutti questi esemplari di compiti devono essere eseguiti, e ora si parlerà del pianificatore.
Pianificatore
Il pianificatore di compiti in Airflow è basato su . Celery è una libreria Python che consente di organizzare una coda, oltre a un'esecuzione asincrona e distribuita di compiti. Dal lato di Airflow, tutti i compiti sono suddivisi in pool. I pool vengono creati manualmente. Di solito, il loro obiettivo è limitare il carico sul lavoro con la sorgente o tipizzare i compiti all'interno del DWH. I pool possono essere gestiti tramite interfaccia web:
Ogni pool ha un limite sul numero di slot. Quando viene creato un DAG, a esso viene assegnato un 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__Il pool definito a livello di DAG può essere sovrascritto a livello di task.
La pianificazione di tutti i task in Airflow è gestita da un processo separato: lo Scheduler. In sostanza, lo Scheduler si occupa dell'intera meccanica di assegnazione dei task all'esecuzione. Un task, prima di essere eseguito, passa attraverso diverse fasi:
- Nel DAG, completati i task precedenti, un nuovo task può essere messo in coda.
- La coda viene ordinata in base alla priorità dei task (le priorità possono essere gestite anch'esse), e, se ci sono slot disponibili nel pool, il task può essere avviato.
- Se c'è un worker celery disponibile, il task viene inviato a lui; inizia il lavoro che hai programmato nel task, utilizzando un determinato operatore.
È piuttosto semplice.
Lo Scheduler opera su molti DAG e su tutte le attività all'interno dei DAG.
Affinché lo Scheduler inizi a lavorare con un DAG, è necessario impostare un programma per il DAG:
dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')Ci sono una serie di preset già pronti: @once, @hourly, @daily, @weekly, @monthly, @yearly.
È anche possibile usare espressioni cron:
dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')Data di esecuzione
Per comprendere come funziona Airflow, è importante capire cosa sia la Data di esecuzione per il DAG. In Airflow, un DAG ha una dimensione Data di esecuzione, ovvero, a seconda del programma di lavoro del DAG, vengono creati istanze di attività per ogni Data di esecuzione. E per ogni Data di esecuzione, le attività possono essere eseguite di nuovo - oppure, ad esempio, un DAG può lavorare contemporaneamente su più Date di esecuzione. Questo è chiaramente mostrato qui:
Sfortunatamente (o forse per fortuna: dipende dalla situazione), se si modifica l'implementazione del task nel DAG, l'esecuzione nelle date di esecuzione precedenti terrà già conto delle correzioni. Questo è utile se si devono ricalcolare i dati nei periodi passati con un nuovo algoritmo, ma negativo perché si perde la riproducibilità del risultato (ovviamente, nessuno impedisce di ripristinare da Git la versione necessaria del sorgente e calcolare una sola volta ciò che serve, come serve).
Generazione di task
L'implementazione del DAG è codice in Python, quindi abbiamo un modo molto comodo per ridurre la quantità di codice quando si lavora, ad esempio, con fonti shardate. Supponiamo di avere come fonte tre shard MySQL, è necessario accedere a ciascuno di essi e prelevare alcuni dati. Inoltre, in modo indipendente e parallelo. Il codice in Python nel DAG può apparire così:
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)Il DAG risulta così:
È possibile aggiungere o rimuovere uno shard semplicemente regolando l'impostazione e aggiornando il DAG. Comodo!
È possibile utilizzare una generazione di codice più complessa, ad esempio lavorare con sorgenti sotto forma di database o descrivere la struttura della tabella. Tenendo conto delle specificità dell'infrastruttura DWH, è possibile generare un processo per caricare N tabelle nel tuo archivio. Oppure, ad esempio, lavorare con un'API che non supporta parametri sotto forma di lista; puoi generare N task in DAG da questa lista, limitare la concorrenza delle richieste nell'API tramite un pool e ottenere i dati necessari dall'API. Flessibile!
Repository
In Airflow c'è un proprio repository backend, un database (può essere MySQL o Postgres, noi usiamo Postgres), in cui vengono memorizzati gli stati delle attività, dei DAG, le impostazioni delle connessioni, le variabili globali, e così via. Qui vorrei dire che il repository in Airflow è molto semplice (circa 20 tabelle) e comodo, se vuoi costruire qualche processo su di esso. Ricordo 100500 tabelle nel repository Informatica, che richiedevano molto tempo per essere comprese prima di riuscire a costruire una query.
Monitoraggio
Considerando la semplicità del repository, puoi costruire un processo di monitoraggio delle attività che sia comodo per te. Noi utilizziamo un notebook in Zeppelin, dove monitoriamo lo stato delle attività:
Questo può essere anche l'interfaccia web di Airflow:
Il codice di Airflow è aperto, quindi abbiamo aggiunto il sistema di alerting su Telegram. Ogni istanza di attività in esecuzione, se si verifica un errore, invia messaggi nel gruppo Telegram, dove si trova l'intero team di sviluppo e supporto.
Otteniamo una risposta rapida tramite Telegram (se necessario), e tramite Zeppelin abbiamo una visione complessiva delle attività in Airflow.
Totale
Airflow è innanzitutto open source e non aspettatevi miracoli. Siate pronti a investire tempo e sforzi per costruire una soluzione funzionante. L'obiettivo è raggiungibile, credetemi, ne vale la pena. La velocità di sviluppo, la flessibilità e la semplicità nell'aggiungere nuovi processi vi piaceranno. Certo, è fondamentale prestare molta attenzione all'organizzazione del progetto e alla stabilità del funzionamento di Airflow: non ci sono miracoli.
Attualmente, il nostro Airflow gestisce quotidianamente circa 6.500 attività. La natura è piuttosto varia. Ci sono attività di caricamento dati nel DWH principale da molte fonti diverse e molto specifiche, ci sono attività di calcolo delle visualizzazioni all'interno del DWH principale, ci sono attività di pubblicazione dei dati in DWH veloce, e molte altre attività ancora — e Airflow le gestisce giorno dopo giorno. Parlando in cifre, si tratta di 2.300 attività ELT di diversa complessità all'interno del DWH (Hadoop), circa 250 fonti di database , un team di 4 sviluppatori ETL, che si dividono tra ETL per il processamento dei dati nel DWH e ELT per il processamento dei dati all'interno del DWH, e naturalmente anche un amministratore, che si occupa dell'infrastruttura del servizio.
Piani per il futuro
Il numero di processi è destinato a crescere, e il nostro principale obiettivo nella parte infrastrutturale di Airflow sarà la scalabilità. Vogliamo costruire un cluster Airflow, dedicare un paio di nodi ai worker di Celery e creare un head duplicated con processi di pianificazione dei job e un repository.
Epilogo
Questo, ovviamente, non è tutto ciò che vorrei dire su Airflow, ma ho cercato di coprire i punti principali. L'appetito viene mangiando, provate e vi piacerà 🙂
Fonte: habr.com
