Nella preparazione dei nostri programmi educativi, ci confrontiamo periodicamente con delle difficoltà relative all'uso di alcuni strumenti. E nel momento in cui ci troviamo a fronteggiarli, non sempre c'è abbastanza documentazione e articoli che possano aiutarci a risolvere questo problema.
È stato così, ad esempio, nel 2015, quando nel programma "Specialista in Big Data" usavamo un cluster Hadoop con Spark per 35 utenti simultanei. Non era chiaro come prepararlo per questo caso d'uso usando YARN. Alla fine, dopo aver approfondito e percorso la strada in autonomia, abbiamo fatto e abbiamo anche partecipato a .
Contesto
Questa volta parleremo di un altro programma – . In questo, i nostri partecipanti costruiscono due tipi di architettura: 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. Lasciate che costruiscano i loro pipeline. Tuttavia, c'è un ma: tutti i nostri programmi sono tecnologici dal punto di vista del processo stesso di apprendimento. Per verificare i laboratori utilizziamo controllori automatici: il partecipante deve accedere al proprio account, premere il pulsante “Verifica” e dopo un po' riceverà un feedback dettagliato su ciò che ha fatto. Ed è proprio in quel momento che iniziamo ad affrontare il nostro problema.
La verifica di questo laboratorio è strutturata in questo modo: inviamo un pacchetto di dati di controllo a Kafka del partecipante, poi Gobblin sposta questo pacchetto di dati su HDFS, quindi Airflow prende questo pacchetto di dati e lo deposita in ClickHouse. La chiave è che Airflow non deve farlo in tempo reale, lo fa secondo un programma: ogni 15 minuti prende un insieme di file e lo carica.
Pertanto, dobbiamo in qualche modo attivare il loro DAG autonomamente su nostra richiesta durante il funzionamento del controllore qui e ora. Cercando su Google, abbiamo scoperto che per le versioni più recenti di Airflow esiste il cosiddetto . La parola experimental, ovviamente, suona inquietante, ma cosa fare… Magari funziona.
Di seguito descriveremo l'intero percorso: dall'installazione di Airflow fino alla generazione di una richiesta POST che attiva il DAG, utilizzando l'API Experimental. Lavoreremo con Ubuntu 16.04.
1. Installazione di Airflow
Verifichiamo che abbiamo Python 3 e virtualenv installati.
$ python3 --version
Python 3.6.6
$ virtualenv --version
15.2.0Se non ne hai uno, installalo.
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) $Installiamo Airflow:
(venv) $ pip install airflowLa versione su cui abbiamo lavorato: 1.10.
Ora dobbiamo creare una cartella airflow_home, dove verranno posizionati 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 crea e inizializza il database del flusso di dati in SQLite:
(venv) $ airflow initdbIl database sarà creato in airflow.db per impostazione predefinita.
Verifichiamo se Airflow è stato installato:
$ airflow version
[2018-11-26 19:38:19,607] {__init__.py:57} INFO - Utilizzo dell'esecutore SequentialExecutor
[2018-11-26 19:38:19,745] {driver.py:123} INFO - Generazione delle tabelle di grammatica da /usr/lib/python3.6/lib2to3/Grammar.txt
[2018-11-26 19:38:19,771] {driver.py:123} INFO - Generazione delle tabelle di grammatica da /usr/lib/python3.6/lib2to3/PatternGrammar.txt
____________ _____________
____ |__( )_________ __/__ /________ __
____ /| |_ /__ ___/_ /_ __ /_ __ _ | /| / /
___ ___ | / _ / _ __/ _ / / /_/ /_ |/ |/ /
_/_/ |_/_/ /_/ /_/ /_/ ____/____/|__/
v1.10.0Se il comando è stato eseguito correttamente, 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 sperimentale
A questo punto Airflow è configurato e pronto per l'uso. Tuttavia, dobbiamo avviare anche l'API sperimentale. I nostri checker sono scritti in Python, quindi tutte le richieste successive saranno fatte utilizzando questa libreria. requests.
In realtà, l'API è già attiva per richieste semplici. Ad esempio, una richiesta del genere permette di testarne il funzionamento:
>>> import requests
>>> host =
>>> airflow_port = 8081 # in questo caso, ma di default è 8080
>>> requests.get('http://{}:{}/{}'.format(host, airflow_port, 'api/experimental/test')).text
'OK'Se hai ricevuto questo messaggio in risposta, significa che tutto funziona.
Tuttavia, quando vorremo attivare il DAG, ci troviamo di fronte al fatto che questo tipo di richiesta non può essere effettuato senza autenticazione.
Per questo sarà necessario eseguire una serie di ulteriori passaggi.
In primo luogo, è necessario aggiungere questo nel file di configurazione:
[api]
auth_backend = airflow.contrib.auth.backends.password_authPoi, è 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": "Creato <DagRun newprolab @ 2019-03-27 10:24:25+00:00: manual__2019-03-27T10:24:25+00:00, esternamente attivato: True>"n}'Richiesta elaborata con successo.
Di conseguenza, diamo un po’ di tempo al DAG per l'elaborazione e facciamo una richiesta alla tabella ClickHouse, cercando di catturare un pacchetto di dati di controllo.
Verifica completata.
Fonte: habr.com
