Oavsett hur mycket teknologin utvecklas, följer alltid en rad förÄldrade tillvÀgagÄngssÀtt med. Detta kan bero pÄ en gradvis övergÄng, den mÀnskliga faktorn, teknologiska krav eller nÄgot annat. Inom databehandling Àr datakÀllor de mest illustrativa i detta avseende. Oavsett hur mycket vi drömmer om att bli av med detta, sÄ lÀnge en del av uppgifterna skickas via meddelandetjÀnster och e-post, för att inte tala om mer arkaiska format. Jag bjuder in till en grundlig genomgÄng av ett av alternativen för Apache Airflow, som illustrerar hur man kan hÀmta data frÄn e-post.
Bakgrund
MÄnga data överförs fortfarande via e-post, frÄn personliga kommunikationer till standarder för interaktion mellan företag. Det Àr bra om man lyckas skriva ett grÀnssnitt för att fÄ data eller sÀtta personer pÄ kontoret som kan föra in denna information i mer anvÀndbara kÀllor, men ofta finns det helt enkelt inte sÄdan möjlighet. Den specifika uppgift jag stötte pÄ var att koppla den vÀlkÀnda CRM-systemet till datalagret, och vidare till OLAP-systemet. Historiskt har vÄrt företag funnit detta system praktiskt i en viss del av verksamheten. DÀrför ville alla ha möjlighet att hantera data Àven frÄn detta externa system. Först och frÀmst undersöktes möjligheten att hÀmta data frÄn det öppna API:et. TyvÀrr tÀckte API:et inte alla nödvÀndiga data, och för att uttrycka det enkelt var det pÄ mÄnga sÀtt bristfÀlligt, och kundsupporten ville inte eller kunde inte ge mer uttömmande funktionalitet. DÀremot möjliggjorde detta system periodisk mottagning av saknade data via e-post som en lÀnk för nedladdning av ett arkiv.
Det Àr viktigt att notera att detta inte var det enda fallet dÀr verksamheten ville samla in data frÄn e-postmeddelanden eller meddelandetjÀnster. I detta fall hade vi dock ingen möjlighet att pÄverka det externa företaget som tillhandahöll vissa data endast pÄ detta sÀtt.
Apache Airflow
För att bygga ETL-processer anvÀnder vi oftast Apache Airflow. För att lÀsaren, som Àr obekant med denna teknologi, bÀttre ska förstÄ hur det ser ut i sitt sammanhang, kommer jag att beskriva ett par inledande aspekter.
Apache Airflow Àr en öppen plattform som anvÀnds för att bygga, köra och övervaka ETL (Extract-Transform-Load) processer i Python. GrundlÀggande koncept inom Airflow Àr riktade acykliska grafer dÀr noderna i grafen representerar specifika processer och kanterna representerar flödet av kontroll eller information. En process kan helt enkelt anropa vilken Python-funktion som helst, eller sÄ kan den ha en mer komplex logik som involverar sekventiell anrop av flera funktioner inom en klass. För de vanligaste operationerna finns det redan mÄnga fÀrdiga lösningar som kan anvÀndas som processer. SÄdana lösningar inkluderar:
- operatörer â för att transportera data frĂ„n en plats till en annan, till exempel frĂ„n en databas till ett datalager;
- sensorer â för att vĂ€nta pĂ„ att ett visst evenemang intrĂ€ffar och styra flödet av kontroll till de följande noderna i grafen;
- hooks â för mer lĂ„gnivĂ„operationer, till exempel för att hĂ€mta data frĂ„n en databas (anvĂ€nds i operatörer);
- och sÄ vidare.
Att beskriva Apache Airflow i detalj i denna artikel vore orealistiskt. Kortare introduktioner kan ses eller .
Hook för att hÀmta data
Först och frÀmst, för att lösa uppgiften mÄste vi skriva en hook, som gör att vi kan:
- ansluta till e-post;
- hitta det nödvÀndiga mailet;
- hÀmta data frÄn mailet.
frÄn airflow.hooks.base_hook import BaseHook
import imaplib
import logging
class IMAPHook(BaseHook):
def __init__(self, imap_conn_id):
"""
IMAP-hook för att hÀmta data frÄn e-post
:param imap_conn_id: E-postanslutningens identifierare
:type imap_conn_id: string
"""
self.connection = self.get_connection(imap_conn_id)
self.mail = None
def authenticate(self):
"""
Ansluter till e-posten
"""
mail = imaplib.IMAP4_SSL(self.connection.host)
response, detail = mail.login(user=self.connection.login, password=self.connection.password)
if response != "OK":
raise AirflowException("Inloggning misslyckades")
else:
self.mail = mail
def get_last_mail(self, check_seen=True, box="INBOX", condition="(UNSEEN)"):
"""
Metod för att hÀmta identifieraren för det senaste mailet,
som uppfyller sökkriterierna
:param check_seen: Markera det senaste mailet som lÀst
:type check_seen: bool
:param box: Namn pÄ lÄdan
:type box: string
:param condition: Villkor för att söka efter mail
:type condition: string
"""
self.authenticate()
self.mail.select(mailbox=box)
response, data = self.mail.search(None, condition)
mail_ids = data[0].split()
logging.info("Följande mail hittades i lÄdan: " + str(mail_ids))
if not mail_ids:
logging.info("Inga nya mail hittades")
return None
mail_id = mail_ids[0]
# om det finns flera sÄdana mail
if len(mail_ids) > 1:
# markera de andra som lÀsta
for id in mail_ids:
self.mail.store(id, "+FLAGS", "\Seen")
# returnera det senaste
mail_id = mail_ids[-1]
# behöver det senaste markeras som lÀst?
if not check_seen:
self.mail.store(mail_id, "-FLAGS", "\Seen")
return mail_idLogiken Àr sÄdan: vi ansluter, hittar det senaste och mest aktuella mailet, om det finns andra - ignorerar vi dem. Denna funktion anvÀnds specifikt, eftersom senare mail innehÄller all information frÄn tidigare. Om sÄ inte Àr fallet, kan man returnera en array med alla mail eller bearbeta det första och de andra vid nÀsta genomgÄng. I allmÀnhet beror allt som vanligt pÄ uppgiften.
Vi lÀgger till tvÄ hjÀlpfunktioner till hooken: en för att ladda ner filen och en för att ladda ner filen frÄn en lÀnk i ett e-postmeddelande. För att nÀmna, de kan flyttas till en operator, vilket beror pÄ hur ofta denna funktion anvÀnds. Vad som ytterligare ska lÀggas till i hooken beror Äterigen pÄ uppgiften: om e-postmeddelandet innehÄller filer kan man ladda ner bilagor, om data kommer i e-postmeddelandet behöver man parsa meddelandet, osv. I mitt fall kommer e-postmeddelandet med en lÀnk till ett arkiv som jag behöver placera pÄ en viss plats och starta den fortsatta bearbetningsprocessen.
def download_from_url(self, url, path, chunk_size=128):
"""
Metod för att ladda ner en fil
:param url: Laddningsadress
:type url: string
:param path: Vart filen ska placeras
:type path: string
:param chunk_size: Hur mÄnga byte som ska skrivas
: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 för att ladda ner en fil frÄn en lÀnk i ett e-postmeddelande
:param mail_id: Identifierare för e-postmeddelandet
:type mail_id: string
:param path: Vart filen ska placeras
: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)Koden Àr enkel, sÄ den behöver knappast ytterligare förklaringar. Jag kommer bara att berÀtta om den magiska raden imap_conn_id. Apache Airflow lagrar anslutningsparametrar (anvÀndarnamn, lösenord, adress och andra parametrar) som kan nÄs via en strÀngidentifierare. Visuellt ser hanteringen av anslutningar ut sÄ hÀr
Sensor för att vÀnta pÄ data
Eftersom vi redan kan ansluta och hÀmta data frÄn e-posten kan vi nu skriva en sensor för att vÀnta pÄ dem. Att skriva en operatör direkt som hanterar data, nÀr de finns, fungerade inte i mitt fall, eftersom andra processer ocksÄ arbetar pÄ grundval av den mottagna datan frÄn e-posten, inklusive de som tar relaterade data frÄn andra kÀllor (API, telefoni, webbanalys etc.). Jag ger ett exempel. Det har dykt upp en ny anvÀndare i CRM-systemet, och vi vet fortfarande inte om deras UUID. DÄ, nÀr vi försöker hÀmta data frÄn SIP-telefoni, kommer vi att fÄ samtal kopplade till deras UUID, men vi kan inte korrekt spara och anvÀnda dem. I sÄdana frÄgor Àr det viktigt att ha i Ätanke dators beroenden, sÀrskilt om de kommer frÄn olika kÀllor. Detta Àr naturligtvis otillrÀckliga ÄtgÀrder för att bevara dataintegriteten, men nödvÀndiga i vissa fall. Och att ockupera resurser utan att anvÀnda dem Àr inte heller rationellt.
SÄledes kommer vÄr sensor att aktivera de efterföljande noderna i grafen om det finns ny information i e-posten, samt mÀrka tidigare information som förÄldrad.
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 TrueHÀmtar och anvÀnder data
För att hĂ€mta och bearbeta data kan man skriva en separat operatör eller anvĂ€nda fĂ€rdiga. Eftersom logiken hittills Ă€r trivial â att hĂ€mta data frĂ„n e-post, föreslĂ„r jag att man anvĂ€nder den standardiserade PythonOperator som exempel.
frÄn airflow.models import DAG
frÄn airflow.operators.python_operator import PythonOperator
frÄn airflow.sensors.my_plugin import MailSensor
frÄn my_plugin.hooks.imap_hook import IMAPHook
start_date = datetime(2020, 4, 4)
# Standardkonfiguration av grafen
args = {
"owner": "exempel",
"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",
)
# Definiera sensorn
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",
)
# Funktion för att hÀmta data frÄn e-post
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("Tom brevlÄda")
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)
# Beskrivning av övriga noder i grafen
...
# SĂ€tt koppling i grafen
mail_check_sensor >> prepare_mail_data
prepare_data >> ...
# Beskrivning av övriga kontrollflödenFörresten, om ditt företags e-post ocksÄ Àr pÄ mail.ru, sÄ kommer du inte kunna söka e-post efter Àmne, avsÀndare, etc. De lovade att införa detta redan 2016, men tydligen har de Àndrat sig. Jag löste problemet genom att skapa en separat mapp för de nödvÀndiga e-postmeddelandena och stÀlla in ett filter i webbgrÀnssnittet för e-posten. PÄ sÄ sÀtt hamnar endast de nödvÀndiga e-postmeddelandena i denna mapp och sökkriterierna blir enklare (UNSEEN).
Sammanfattningsvis har vi följande sekvens: kontrollerar om det finns nya e-postmeddelanden som matchar kriterierna, om det finns, laddar vi ner arkivet via lÀnken i det senaste e-postmeddelandet.
Under de sista ellipserna har vi utelÀmnat att detta arkiv kommer att packas upp, datan frÄn arkivet kommer att rensas och bearbetas, och i slutÀndan kommer allt detta att gÄ vidare till ETL-processens konveyör, men det ligger utanför Àmnet för artikeln. Om det kÀnns intressant och anvÀndbart, fortsÀtter jag gÀrna att beskriva ETL-lösningar och dess delar för Apache Airflow.
KĂ€lla: habr.com
