Durante la preparazione dei nostri programmi educativi, ci troviamo periodicamente di fronte a difficoltà relative all'uso di alcuni strumenti. E nel momento in cui ci imbattiamo in essi, non sempre c'è abbastanza documentazione o articoli che possano aiutare a risolvere questo problema.
È stato così, ad esempio, nel 2015, quando nel programma "Specialista in Big Data" utilizzavamo un cluster Hadoop con Spark per 35 utenti simultanei. Non era chiaro come prepararlo per un tale caso d'uso utilizzando YARN. Alla fine, dopo averlo capito e avendo fatto il percorso da soli, abbiamo realizzato e abbiamo anche partecipato a .
Antefatti
Questa volta parleremo di un altro programma - . In esso, i nostri partecipanti costruiscono due tipi di architetture: lambda e kappa. Nella architettura lambda, nell'ambito dell'elaborazione batch, utilizziamo Airflow per spostare i log da HDFS a ClickHouse.
In generale, va tutto bene. Lasciamo che costruiscano le loro pipeline. Tuttavia, c'è un problema: tutti i nostri programmi sono tecnologici dal punto di vista del processo di apprendimento. Per controllare i laboratori, utilizziamo checker automatici: il partecipante deve entrare nel proprio account, premere il pulsante “Controlla” e dopo un certo tempo vede un feedback dettagliato su ciò che ha fatto. Ed è proprio in quel momento che iniziamo ad affrontare il nostro problema.
Il controllo di questo laboratorio è strutturato in questo modo: inviamo un pacchetto di dati di controllo in Kafka all'utente, successivamente Gobblin sposta questo pacchetto di dati su HDFS, poi Airflow prende questo pacchetto di dati e lo inserisce in ClickHouse. Il punto è che Airflow non deve farlo in tempo reale, lo fa secondo un programma: ogni 15 minuti prende un insieme di file e li carica.
A questo punto, abbiamo bisogno di attivare il loro DAG autonomamente su richiesta durante il lavoro del checker qui e ora. Cercando su Google, abbiamo scoperto che per le versioni più recenti di Airflow esiste quello che si chiama . La parola experimental, ovviamente, suona minacciosa, ma che fare… Forse funzionerà.
Sotto descriveremo tutto il percorso: dall'installazione di Airflow alla formazione della richiesta POST che attiva il DAG, utilizzando l'Experimental API. Lavoreremo con Ubuntu 16.04.
1. Installazione di Airflow
Controlliamo che abbiamo Python 3 e virtualenv installati.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Se non avete nulla di tutto ciò, installatelo.
Ora creiamo una cartella in cui lavoreremo ulteriormente con Airflow.
$ mkdir
$ cd /path/to/your/new/directory
$ virtualenv -p which python3 venv
$ source venv/bin/activate
(venv) $Installdiamo Airflow:
(venv) $ pip install airflowVersione su cui abbiamo lavorato: 1.10.
Ora dobbiamo creare una cartella airflow_home, dove saranno collocati i file DAG e i plugin di Airflow. Dopo aver creato la cartella, imposteremo la variabile di ambiente AIRFLOW_HOME.
(venv) $ cd /path/to/my/airflow/workspace
(venv) $ mkdir airflow_home
(venv) $ export AIRFLOW_HOME=Il passo successivo è eseguire il comando che crearà e inizializzerà il database del flusso di dati in SQLite:
(venv) $ airflow initdbIl database sarà creato in airflow.db per impostazione predefinita.
Controlliamo se Airflow si è installato:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Using executor SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generating grammar tables from /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / /
___ ___ | / _ / _ __/ _ / /_ /_ /| /|/ /
_/_/_ |_/_/_ /_/_/ /_/_/ /_/_/ ____/____/|__/
v1.10.0Se il comando è andato a buon fine, Airflow ha creato il suo file di configurazione airflow.cfg in AIRFLOW_HOME:
$ tree
.
├── airflow.cfg
└── unittests.cfgAirflow ha un'interfaccia web. Può essere avviata eseguendo il comando:
(venv) $ airflow webserver --port 8081Ora puoi accedere all'interfaccia web nel browser sulla porta 8081 sull'host dove Airflow è stato avviato, ad esempio: <hostname:8081>.
2. Lavorare con l'API Experimental
A questo punto Airflow è configurato e pronto per l'uso. Tuttavia, dobbiamo avviare anche l'API Experimental. I nostri checker sono scritti in Python, quindi successivamente tutte le richieste saranno in esso utilizzando la libreria requests.
In realtà l'API è già in funzione per richieste semplici. Ad esempio, questa richiesta permette di testarne il funzionamento:
>>> import requests
>>> host =
>>> airflow_port = 8081 #nel nostro caso questo, di default è 8080
>>> requests.get('http://{}:{} /{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'Se hai ricevuto questo messaggio come risposta, significa che tutto funziona.
Tuttavia, quando desideriamo attivare il DAG, ci troveremo di fronte al fatto che questo tipo di richiesta non può essere effettuato senza autenticazione.
Per questo sarà necessario eseguire un certo numero di ulteriori azioni.
Innanzitutto, nel file di configurazione bisogna aggiungere questo:
[api]
auth_backend = airflow.contrib.auth.backends.password_authSuccessivamente, è necessario creare un utente con diritti di amministratore:
>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.Admin())
>>> user.username = 'new_user_name'
>>> user.password = 'set_the_password'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()Poi, è necessario creare un utente con diritti normali, a cui sarà consentito attivare il DAG.
>>> import airflow
>>> from airflow import models, settings
>>> from airflow.contrib.auth.backends.password_auth import PasswordUser
>>> user = PasswordUser(models.User())
>>> user.username = 'newprolab'
>>> user.password = 'Newprolab2019!'
>>> session = settings.Session()
>>> session.add(user)
>>> session.commit()
>>> session.close()
>>> exit()Ora tutto è pronto.
3. Avvio della richiesta POST
La richiesta POST apparirà così:
>>> dag_id = newprolab
>>> url = 'http://{}:{}{/}{}/{}'.format(host, airflow_port, 'api/experimental/dags', dag_id, 'dag_runs')
>>> data = {"conf":"{"key":"value"}"}
>>> headers = {'Content-type': 'application/json'}
>>> auth = ('newprolab', 'Newprolab2019!')
>>> uri = requests.post(url, data=json.dumps(data), headers=headers, auth=auth)
>>> uri.text
'{n "message": "Created "n}n'La richiesta è stata elaborata con successo.
Di conseguenza, diamo un po' di tempo al DAG per elaborare e facciamo una richiesta alla tabella ClickHouse, cercando di catturare il pacchetto di dati di controllo.
Controllo completato.
Fonte: habr.com
