Nonostante il continuo avanzamento della tecnologia, rimangono sempre approcci obsoleti a seguirne l'evoluzione. Questo può essere dovuto a una transizione graduale, al fattore umano, alle necessità tecnologiche o ad altro. Nel campo dell'elaborazione dei dati, le fonti di dati sono particolarmente rappresentative in questo contesto. Non importa quanto desideriamo liberarci di questo, finché parte dei dati viene inviato tramite messaggi e email, senza contare formati ancora più arcaici. Invito a scoprire sotto un esempio per Apache Airflow, che illustra come estrarre dati dalle email.
Antefatti
Molti dati vengono ancora trasmessi tramite email, dalle comunicazioni interpersonali agli standard di interazione tra le aziende. È positivo riuscire a scrivere un'interfaccia per ricevere i dati o avere persone in ufficio che inseriscano queste informazioni in fonti più utili, ma spesso tale possibilità non è disponibile. Il compito specifico con cui mi sono confrontato è stato quello di connettere un noto sistema CRM a un data warehouse, e successivamente a un sistema OLAP. Storicamente, per la nostra azienda, l'uso di questo sistema è stato comodo in un'area specifica del business. Perciò c'era un forte desiderio di poter gestire i dati anche da questo sistema esterno. Prima di tutto, abbiamo esaminato la possibilità di ottenere dati tramite un API aperta. Purtroppo, l'API non copriva l'ottenimento di tutti i dati necessari e, parlando in termini semplici, presentava molti difetti, mentre il supporto tecnico non ha voluto o potuto concedere funzionalità più complete. Tuttavia, questo sistema offriva la possibilità di ricevere periodicamente i dati mancanti via email sotto forma di link per il download dell'archivio.
È importante notare che questo non era l'unico caso in cui il business voleva raccogliere dati da email o messaggeri. Tuttavia, in questo caso non abbiamo potuto influenzare l'azienda esterna che forniva parte dei dati solo in questo modo.
Apache Airflow
Per costruire processi ETL utilizziamo più frequentemente Apache Airflow. Affinché il lettore che non conosce questa tecnologia possa comprendere meglio come si presenta nel contesto e nel complesso, descriverò un paio di introduzioni.
Apache Airflow è una piattaforma open-source utilizzata per costruire, eseguire e monitorare processi ETL (Extract-Transform-Loading) in Python. Il concetto principale in Airflow è il grafo orientato aciclico, dove i nodi del grafo rappresentano processi specifici e gli archi del grafo rappresentano il flusso di controllo o di informazioni. Un processo può semplicemente chiamare qualsiasi funzione Python, oppure può avere una logica più complessa data dalla chiamata sequenziale di più funzioni all'interno di una classe. Per le operazioni più comuni esistono già molte soluzioni pronte all'uso che possono essere utilizzate come processi. Tra queste soluzioni ci sono:
- operatori — per trasferire dati da un luogo all'altro, ad esempio da una tabella di un database a un data warehouse;
- sensori — per attendere il verificarsi di un determinato evento e dirigere il flusso di controllo verso i successivi nodi del grafo;
- hook — per operazioni a basso livello, ad esempio, per ottenere dati da una tabella di un database (utilizzati negli operatori);
- ecc.
Descrivere Apache Airflow in dettaglio in questo articolo non sarebbe utile. È possibile dare un'occhiata a brevi introduzioni o .
Hook per ottenere dati
In primo luogo, per risolvere il problema dobbiamo scrivere un hook che ci permetta di:
- collegarci all'email;
- trovare l'email desiderata;
- ottenere dati dall'email.
da airflow.hooks.base_hook import BaseHook
import imaplib
import logging
class IMAPHook(BaseHook):
def __init__(self, imap_conn_id):
"""
IMAP hook per ottenere dati dalla posta elettronica
:param imap_conn_id: Identificatore della connessione alla posta
:type imap_conn_id: string
"""
self.connection = self.get_connection(imap_conn_id)
self.mail = None
def authenticate(self):
"""
Ci connettiamo alla posta
"""
mail = imaplib.IMAP4_SSL(self.connection.host)
response, detail = mail.login(user=self.connection.login, password=self.connection.password)
if response != "OK":
raise AirflowException("Accesso non riuscito")
else:
self.mail = mail
def get_last_mail(self, check_seen=True, box="INBOX", condition="(UNSEEN)"):
"""
Metodo per ottenere l'identificatore dell'ultima email,
che soddisfa i criteri di ricerca
:param check_seen: Contrassegnare l'ultima email come letta
:type check_seen: bool
:param box: Nome della casella
:type box: string
:param condition: Criteri di ricerca delle email
:type condition: string
"""
self.authenticate()
self.mail.select(mailbox=box)
response, data = self.mail.search(None, condition)
mail_ids = data[0].split()
logging.info("Nella casella sono state trovate le seguenti email: " + str(mail_ids))
if not mail_ids:
logging.info("Non sono state trovate email nuove")
return None
mail_id = mail_ids[0]
# se ci sono più email
if len(mail_ids) > 1:
# contrassegniamo le altre come lette
for id in mail_ids:
self.mail.store(id, "+FLAGS", "\Seen")
# restituiamo l'ultima
mail_id = mail_ids[-1]
# è necessario contrassegnare l'ultima come letta?
if not check_seen:
self.mail.store(mail_id, "-FLAGS", "\Seen")
return mail_idLa logica è questa: ci connettiamo, troviamo l'ultima email più recente, e se ci sono altre, le ignoriamo. Si utilizza proprio questa funzione, perché le email più recenti contengono tutti i dati delle precedenti. Se non è così, si può restituire un array di tutte le email o elaborare la prima, mentre le altre alla prossima iterazione. In generale, tutto dipende sempre dal compito.
Aggiungiamo al gancio due funzioni ausiliarie: per scaricare un file e per scaricare un file tramite un link nell'email. A proposito, possono essere estratte in un'operazione, a seconda della frequenza di utilizzo di questa funzionalità. Cosa altro aggiungere al gancio, dipende nuovamente dal compito: se nell'email arrivano subito file, si possono scaricare gli allegati all'email; se i dati arrivano nell'email, si deve analizzare l'email e così via. Nel mio caso, l'email arriva con un link a un archivio che devo posizionare in un posto specifico e avviare il processo successivo di elaborazione.
def download_from_url(self, url, path, chunk_size=128):
"""
Metodo per scaricare un file
:param url: Indirizzo di download
:type url: string
:param path: Dove posizionare il file
:type path: string
:param chunk_size: Quanti byte scrivere
:type chunk_size: int
"""
r = requests.get(url, stream=True)
with open(path, "wb") as fd:
for chunk in r.iter_content(chunk_size=chunk_size):
fd.write(chunk)
def download_mail_href_attachment(self, mail_id, path):
"""
Metodo per scaricare un file tramite un link nell'email
:param mail_id: Identificatore dell'email
:type mail_id: string
:param path: Dove posizionare il file
:type path: string
"""
response, data = self.mail.fetch(mail_id, "(RFC822)")
raw_email = data[0][1]
raw_soup = raw_email.decode().replace("r", "").replace("n", "")
parse_soup = BeautifulSoup(raw_soup, "html.parser")
link_text = ""
for a in parse_soup.find_all("a", href=True, text=True):
link_text = a["href"]
self.download_from_url(link_text, path)Il codice è semplice, quindi difficilmente necessita di ulteriori spiegazioni. Parlerò solo della riga magica imap_conn_id. Apache Airflow memorizza i parametri di connessione (nome utente, password, indirizzo e altri parametri), a cui è possibile accedere tramite un identificatore di stringa. Visivamente, la gestione delle connessioni appare così:
Sensore per attendere i dati
Poiché possiamo già connetterci e ricevere dati dalle email, ora possiamo scrivere un sensore per attenderli. Scrivere subito un operatore che elabori i dati, se presenti, non è stato possibile nel mio caso, poiché a partire dai dati ricevuti dalle email funzionano anche altri processi, tra cui quelli che prelevano dati correlati da fonti diverse (API, telefonia, web metrics, ecc.). Faccio un esempio. In un sistema CRM è apparso un nuovo utente e non sappiamo ancora il suo UUID. Quindi, quando proviamo a ottenere dati dalla telefonia SIP, riceviamo chiamate collegate al suo UUID, ma non possiamo conservarle e utilizzarle correttamente. In questi casi è importante tenere presente la dipendenza dei dati, specialmente se provengono da fonti diverse. Questo, ovviamente, non è un modo sufficiente per garantire l'integrità dei dati, ma in alcuni casi è necessario. E occupare risorse inutilmente non è neppure razionale.
In questo modo, il nostro sensore avvierà i successivi vertici del grafo, se ci sono nuove informazioni nella posta, e contrassegnerà le informazioni precedenti come non attuali.
from airflow.sensors.base_sensor_operator import BaseSensorOperator
from airflow.utils.decorators import apply_defaults
from my_plugin.hooks.imap_hook import IMAPHook
class MailSensor(BaseSensorOperator):
@apply_defaults
def __init__(self, conn_id, check_seen=True, box="Inbox", condition="(UNSEEN)", *args, **kwargs):
super().__init__(*args, **kwargs)
self.conn_id = conn_id
self.check_seen = check_seen
self.box = box
self.condition = condition
def poke(self, context):
conn = IMAPHook(self.conn_id)
mail_id = conn.get_last_mail(check_seen=self.check_seen, box=self.box, condition=self.condition)
if mail_id is None:
return False
else:
return TrueOtteniamo e utilizziamo i dati
Per ottenere e elaborare i dati si può scrivere un operatore dedicato o utilizzare quelli già pronti. Poiché la logica è al momento semplice — prelevare i dati da un email — per esempio suggerisco il PythonOperator standard.
da airflow.models importare DAG
da airflow.operators.python_operator importare PythonOperator
da airflow.sensors.my_plugin importare MailSensor
da my_plugin.hooks.imap_hook importare IMAPHook
start_date = datetime(2020, 4, 4)
# Configurazione standard del grafo
args = {
"owner": "example",
"start_date": start_date,
"email": ["home@home.ru"],
"email_on_failure": False,
"email_on_retry": False,
"retry_delay": timedelta(minutes=15),
"provide_context": False,
}
dag = DAG(
dag_id="test_etl",
default_args=args,
schedule_interval="@hourly",
)
# Definiamo il sensore
mail_check_sensor = MailSensor(
task_id="check_new_emails",
poke_interval=10,
conn_id="mail_conn_id",
timeout=10,
soft_fail=True,
box="my_box",
dag=dag,
mode="poke",
)
# Funzione per ottenere i dati dalla e-mail
def prepare_mail():
imap_hook = IMAPHook("mail_conn_id")
mail_id = imap_hook.get_last_mail(check_seen=True, box="my_box")
if mail_id is None:
raise AirflowException("Mail box vuota")
conn.download_mail_href_attachment(mail_id, "./path.zip")
prepare_mail_data = PythonOperator(task_id="prepare_mail_data", default_args=args, dag=dag, python_callable= prepare_mail)
# Descrizione degli altri nodi del grafo
...
# Creiamo connessione nel grafo
mail_check_sensor >> prepare_mail_data
prepare_data >> ...
# Descrizione degli altri flussi di controlloA proposito, se la tua e-mail aziendale è anche su mail.ru, non avrai accesso alla ricerca di e-mail per oggetto, mittente e così via. Promettevano di introdurla già nel lontano 2016, ma evidentemente hanno cambiato idea. Ho risolto questo problema creando una cartella separata per le e-mail necessarie e impostando un filtro nell'interfaccia web della posta per le e-mail desiderate. In questo modo, solo le e-mail necessarie finiscono in questa cartella e le condizioni di ricerca nel mio caso sono semplicemente (UNSEEN).
In sintesi, abbiamo la seguente sequenza: controlliamo se ci sono nuove e-mail che soddisfano i criteri, se ce ne sono, scarichiamo l'archivio dal link dell'ultima e-mail.
Con le ultime ellissi si omette che questo archivio sarà estratto, i dati dell'archivio saranno puliti e elaborati, e alla fine tutto ciò andrà avanti nel processo ETL, ma questo esula già dal tema dell'articolo. Se ti è sembrato interessante e utile, sarò felice di continuare a descrivere le soluzioni ETL e le loro parti per Apache Airflow.
Fonte: habr.com
