Ne glede na to, kako se tehnologije razvijajo, za razvojem vedno sledi vrsta zastarelih pristopov. To je lahko posledica postopnega prehoda, človeškega dejavnika, tehnoloških potreb ali česa drugega. Na področju obdelave podatkov so najbolj značilni v tej kategoriji viri podatkov. Kako zelo si želimo, da bi se tega znebili, dokler se del podatkov pošilja preko sporočil in elektronskih pisem, ne govorimo že o bolj archaicnih formatih. Vabim vas, da se pod kat razpravimo o eni od možnosti za Apache Airflow, ki prikazuje, kako lahko pridobimo podatke iz elektronskih pisem.
Pozadina
Številni podatki se še vedno prenašajo preko elektronske pošte, od medosebne komunikacije do standardov med podjetji. Dobro je, če lahko za pridobivanje podatkov napišemo vmesnik ali zaposlene v pisarni, ki bodo te informacije vnašali v bolj priročne vire, vendar take možnosti pogosto preprosto ni. Konkreten izziv, s katerim sem se soočil, je bilo povezovanje znane CRM sistema s podatkovnim skladiščem, nato pa z OLAP sistemom. Tako je zgodovinsko izšlo, da je bilo za naše podjetje uporabo te sistema primerno v določenem segmentu poslovanja. Zato je bilo vsem zelo pomembno, da lahko upravljamo s podatki tudi iz tega zunanjega sistema. Najprej smo seveda preučili možnost pridobivanja podatkov iz odprtega API-ja. Na žalost API ni pokrival pridobivanja vseh potrebnih podatkov in, preprosto povedano, je bil v marsičem pomanjkljiv, tehnična podpora pa ni želela ali uspela ugoditi in ponuditi bolj izčrpen funkcionalnost. Vendar pa je ta sistem omogočil občasno prejemanje manjkajočih podatkov na e-pošto v obliki povezave za prenos arhiva.
Treba je omeniti, da to ni bil edini primer, kjer je podjetje želelo zbirati podatke iz elektronskih sporočil ali sporočil. Vendar pa v tem primeru nismo mogli vplivati na zunanje podjetje, ki zagotavlja del podatkov samo na ta način.
Apache Airflow
Za izgradnju ETL procesa najčešće koristimo Apache Airflow. Da bi čitatelj koji nije upoznat s ovom tehnologijom bolje shvatio kako to izgleda u kontekstu, opisat ću nekoliko uvodnih informacija.
Apache Airflow je slobodna platforma koja se koristi za izgradnju, izvođenje i nadzor ETL (Extract-Transform-Loading) procesa na Pythonu. Osnovni koncept u Airflow-u je usmjereni aciklični graf, gdje su vrhovi grafa konkretni procesi, a rubovi grafa su tok upravljanja ili informacija. Proces može jednostavno pozvati bilo koju Python funkciju, ili imati složeniju logiku koja uključuje sekvencijalno pozivanje više funkcija unutar konteksta klase. Već postoje mnogi gotovi resursi za najčešće operacije koje se mogu koristiti kao procesi. Takvi resursi uključuju:
- operatore — za prijenos podataka s jednog mjesta na drugo, na primjer, iz tablice baze podataka u skladište podataka;
- senzore — za čekanje na određeni događaj i preusmjeravanje toka upravljanja u sljedeće vrhove grafa;
- kukice — za niskonivo operacije, na primjer, za dobivanje podataka iz tablice baze podataka (koriste se u operatorima);
- itd.
Deteljno opisivanje Apache Airflow-a u ovom članku ne bi bilo korisno. Kratka uvodna objašnjenja mogu se pogledati ili .
Kukica za dobivanje podataka
Prvo, da bismo riješili zadatak, potrebno je napisati kukicu koja bi nam omogućila:
- spojiti se na e-mail;
- pronaći potrebnu poruku;
- dobiti podatke iz poruke.
from airflow.hooks.base_hook import BaseHook
import imaplib
import logging
class IMAPHook(BaseHook):
def __init__(self, imap_conn_id):
"""
IMAP hook for retrieving data from email
:param imap_conn_id: Email connection identifier
:type imap_conn_id: string
"""
self.connection = self.get_connection(imap_conn_id)
self.mail = None
def authenticate(self):
"""
Connecting to the email
"""
mail = imaplib.IMAP4_SSL(self.connection.host)
response, detail = mail.login(user=self.connection.login, password=self.connection.password)
if response != "OK":
raise AirflowException("Sign in failed")
else:
self.mail = mail
def get_last_mail(self, check_seen=True, box="INBOX", condition="(UNSEEN)"):
"""
Method to retrieve the ID of the last email,
fulfilling the search criteria
:param check_seen: Mark the last email as read
:type check_seen: bool
:param box: Mailbox name
:type box: string
:param condition: Email search conditions
:type condition: string
"""
self.authenticate()
self.mail.select(mailbox=box)
response, data = self.mail.search(None, condition)
mail_ids = data[0].split()
logging.info("The following emails were found in the mailbox: " + str(mail_ids))
if not mail_ids:
logging.info("No new emails found")
return None
mail_id = mail_ids[0]
# if there are multiple such emails
if len(mail_ids) > 1:
# mark the others as read
for id in mail_ids:
self.mail.store(id, "+FLAGS", "\Seen")
# return the last one
mail_id = mail_ids[-1]
# should we mark the last as read
if not check_seen:
self.mail.store(mail_id, "-FLAGS", "\Seen")
return mail_idThe logic is as follows: we connect, find the most recent email, and if there are others, we ignore them. This function is used specifically because later emails contain all the information from earlier ones. If that is not the case, we can return an array of all emails or handle the first one and the others in the next pass. Ultimately, as always, it depends on the task.
Dodajemo dve pomoćne funkcije uz hook: za preuzimanje datoteke i za preuzimanje datoteke putem linka iz e-pošte. Usput rečeno, može ih se izdvojiti u operator, što zavisi od učestalosti korišćenja ove funkcionalnosti. Šta još dodati u hook, opet zavisi od zadatka: ako u e-pošti dolaze odmah datoteke, možemo preuzeti priloge, ako podaci dolaze u e-pošti, onda treba analizirati e-poštu itd. U mom slučaju, e-pošta dolazi s jednim linkom ka arhivi, koju trebam staviti na određeno mesto i pokrenuti dalje procesiranje.
def download_from_url(self, url, path, chunk_size=128):
"""
Metod za preuzimanje datoteke
:param url: Adresa za preuzimanje
:type url: string
:param path: Gde staviti datoteku
:type path: string
:param chunk_size: Koliko bajtova pisati
: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):
"""
Metod za preuzimanje datoteke putem linka iz e-pošte
:param mail_id: Identifikator e-pošte
:type mail_id: string
:param path: Gde staviti datoteku
: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)Kod je jednostavan, tako da verovatno ne zahteva dodatna objašnjenja. Pričaću samo o magičnoj liniji imap_conn_id. Apache Airflow čuva parametre konekcija (korisničko ime, lozinka, adresa i drugi parametri), kojima se može pristupiti putem string identifikatora. Vizuelno upravljanje konekcijama izgleda ovako
Senzor za čekanje podataka
Ker smo se že naučili povezovati in pridobivati podatke iz e-pošte, zdaj lahko napišemo senzor, ki bo čakal na te podatke. Takojšnjega operaterja, ki bi obdeloval podatke, če so na voljo, nisem uspel napisati, saj na podlagi pridobljenih podatkov iz e-pošte delujejo tudi drugi procesi, vključno s tistimi, ki pridobivajo povezane podatke iz drugih virov (API, telefonija, spletna meritev itd.). Tukaj je primer. V CRM sistemu je nastavil nov uporabnik, in še ne vemo za njegov UUID. Tako bomo ob poskusu pridobivanja podatkov iz SIP telefonije dobili klice, povezane z njegovim UUID, a jih ne bomo mogli pravilno shraniti in uporabiti. Pri takih vprašanjih je pomembno upoštevati odvisnost podatkov, še posebej, če prihajajo iz različnih virov. To so seveda nezadostni ukrepi za ohranjanje celovitosti podatkov, vendar so v nekaterih primerih nujni. Zasedanje virov brez razloga prav tako ni racionalno.
Tako bo naš senzor sprožil naslednje vrhove grafa, če so na voljo sveže informacije v e-pošti, hkrati pa bo označil prejšnje informacije kot zastarele.
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 TruePridobivamo in uporabljamo podatke
Za pridobivanje in obdelavo podatkov lahko napišete ločen operater, lahko pa uporabite tudi obstoječe. Ker je trenutno logika trivialna — zajeti podatke iz e-pošte, za primer predlagam standardni PythonOperator.
from airflow.models import DAG
from airflow.operators.python_operator import PythonOperator
from airflow.sensors.my_plugin import MailSensor
from my_plugin.hooks.imap_hook import IMAPHook
start_date = datetime(2020, 4, 4)
# Standard configuration of the graph
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",
)
# Defining the sensor
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",
)
# Function to retrieve data from the email
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("Empty mailbox")
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)
# Description of other nodes in the graph
...
# Defining the connection in the graph
mail_check_sensor >> prepare_mail_data
prepare_data >> ...
# Description of other control flowsBy the way, if your corporate email is also on mail.ru, you won't be able to search emails by subject, sender, etc. They promised to introduce this back in 2016, but it seems they changed their minds. I solved this problem by creating a separate folder for the necessary emails and setting up a filter in the web interface for the needed emails. Thus, only the necessary emails go into this folder and the search conditions in my case are simply (UNSEEN).
In summary, we have the following sequence: we check for new emails that meet the criteria, if there are any, we download the archive from the link in the last email.
The last ellipses imply that this archive will be unpacked, data from the archive will be cleaned and processed, and ultimately all of this will continue on the ETL process conveyor, but this goes beyond the scope of the article. If you found it interesting and useful, I would be happy to continue describing ETL solutions and their parts for Apache Airflow.
Izvor: habr.com
