Bez ohľadu na to, do akej miery sa technológia vyvíja, za vývojom vždy stojí rad zastaraných prístupov. Môže to byť spôsobené hladkým prechodom, ľudskými faktormi, technologickými potrebami alebo niečím iným. V oblasti spracovania dát sú v tejto časti najodhaľujúce zdroje dát. Bez ohľadu na to, ako veľmi snívame o tom, že sa toho zbavíme, zatiaľ sa časť údajov posiela v instant messengeroch a e-mailoch, nehovoriac o archaickejších formátoch. Pozývam vás, aby ste rozobrali jednu z možností pre Apache Airflow, ktorá ilustruje, ako môžete získavať údaje z e-mailov.
pravek
Veľa údajov sa stále prenáša prostredníctvom e-mailu, od medziľudskej komunikácie až po štandardy interakcie medzi spoločnosťami. Je dobré, ak je možné napísať rozhranie na získavanie údajov alebo dať do kancelárie ľudí, ktorí budú tieto informácie vkladať do pohodlnejších zdrojov, ale často to jednoducho nie je možné. Špecifická úloha, pred ktorou som stál, bolo prepojenie notoricky známeho CRM systému s dátovým skladom a následne so systémom OLAP. Historicky sa stalo, že pre našu spoločnosť bolo použitie tohto systému výhodné v konkrétnej oblasti podnikania. Preto naozaj každý chcel vedieť operovať aj s dátami z tohto systému tretích strán. V prvom rade sa samozrejme skúmala možnosť získavania údajov z otvoreného API. Žiaľ, API nepokrývalo získavanie všetkých potrebných údajov a, zjednodušene povedané, bolo v mnohých smeroch pokrivené a technická podpora sa nechcela alebo nemohla stretnúť na polceste, aby poskytla komplexnejšiu funkcionalitu. Tento systém však poskytoval príležitosť pravidelne dostávať chýbajúce údaje poštou vo forme odkazu na stiahnutie archívu.
Treba podotknúť, že to nebol jediný prípad, kedy chcel biznis zbierať dáta z emailov alebo instant messengerov. V tomto prípade sme však nemohli ovplyvniť tretiu spoločnosť, ktorá poskytuje časť údajov iba týmto spôsobom.
prúdenie vzduchu apache
Na budovanie ETL procesov najčastejšie využívame Apache Airflow. Aby čitateľ, ktorý túto technológiu nepozná, lepšie pochopil, ako vyzerá v kontexte a celkovo, popíšem pár úvodných.
Apache Airflow je bezplatná platforma, ktorá sa používa na vytváranie, spúšťanie a monitorovanie procesov ETL (Extract-Transform-Loading) v Pythone. Hlavným konceptom v Airflow je riadený acyklický graf, kde vrcholy grafu sú špecifické procesy a okraje grafu sú tok riadenia alebo informácií. Proces môže jednoducho volať akúkoľvek funkciu Pythonu alebo môže mať zložitejšiu logiku z postupného volania niekoľkých funkcií v kontexte triedy. Pre najčastejšie operácie už existuje veľa hotových vývojových riešení, ktoré možno použiť ako procesy. Takýto vývoj zahŕňa:
- operátori - na prenos údajov z jedného miesta na druhé, napríklad z databázovej tabuľky do dátového skladu;
- senzory - na čakanie na výskyt určitej udalosti a nasmerovanie toku riadenia na nasledujúce vrcholy grafu;
- háčiky - pre operácie nižšej úrovne, napríklad na získanie údajov z databázovej tabuľky (používané v príkazoch);
- atď
Bolo by nevhodné podrobne popisovať Apache Airflow v tomto článku. Môžete si pozrieť krátke úvody alebo .
Háčik na získavanie údajov
Najprv musíme na vyriešenie problému napísať háčik, pomocou ktorého by sme mohli:
- pripojiť k e-mailu
- nájsť správne písmeno
- prijímať údaje z listu.
from airflow.hooks.base_hook import BaseHook
import imaplib
import logging
class IMAPHook(BaseHook):
def __init__(self, imap_conn_id):
"""
IMAP hook для получения данных с электронной почты
:param imap_conn_id: Идентификатор подключения к почте
:type imap_conn_id: string
"""
self.connection = self.get_connection(imap_conn_id)
self.mail = None
def authenticate(self):
"""
Подключаемся к почте
"""
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)"):
"""
Метод для получения идентификатора последнего письма,
удовлетвораяющего условиям поиска
:param check_seen: Отмечать последнее письмо как прочитанное
:type check_seen: bool
:param box: Наименования ящика
:type box: string
:param condition: Условия поиска писем
:type condition: string
"""
self.authenticate()
self.mail.select(mailbox=box)
response, data = self.mail.search(None, condition)
mail_ids = data[0].split()
logging.info("В ящике найдены следующие письма: " + str(mail_ids))
if not mail_ids:
logging.info("Не найдено новых писем")
return None
mail_id = mail_ids[0]
# если таких писем несколько
if len(mail_ids) > 1:
# отмечаем остальные прочитанными
for id in mail_ids:
self.mail.store(id, "+FLAGS", "\Seen")
# возвращаем последнее
mail_id = mail_ids[-1]
# нужно ли отметить последнее прочитанным
if not check_seen:
self.mail.store(mail_id, "-FLAGS", "\Seen")
return mail_idLogika je takáto: spojíme sa, nájdeme posledné najrelevantnejšie písmeno, ak existujú ďalšie, ignorujeme ich. Táto funkcia sa používa, pretože neskoršie písmená obsahujú všetky údaje predchádzajúcich. Ak to tak nie je, môžete vrátiť pole všetkých písmen alebo spracovať prvé a ostatné pri ďalšom prechode. Vo všeobecnosti všetko, ako vždy, závisí od úlohy.
K háku pridávame dve pomocné funkcie: na stiahnutie súboru a na stiahnutie súboru pomocou odkazu z emailu. Mimochodom, dajú sa presunúť k operátorovi, záleží na frekvencii využívania tejto funkcionality. Čo ešte pridať k háku, opäť závisí od úlohy: ak sú súbory okamžite prijaté v liste, potom si môžete stiahnuť prílohy k listu, ak sú údaje prijaté v liste, musíte list analyzovať, atď. V mojom prípade list prichádza s jedným odkazom na archív, ktorý potrebujem umiestniť na určité miesto a spustiť ďalší proces spracovania.
def download_from_url(self, url, path, chunk_size=128):
"""
Метод для скачивания файла
:param url: Адрес загрузки
:type url: string
:param path: Куда положить файл
:type path: string
:param chunk_size: По сколько байтов писать
: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):
"""
Метод для скачивания файла по ссылке из письма
:param mail_id: Идентификатор письма
:type mail_id: string
:param path: Куда положить файл
: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)Kód je jednoduchý, takže nepotrebuje ďalšie vysvetlenie. Poviem vám len o magickej čiare imap_conn_id. Apache Airflow ukladá parametre pripojenia (login, heslo, adresa a ďalšie parametre), ku ktorým možno pristupovať pomocou identifikátora reťazca. Vizuálne správa pripojenia vyzerá takto
Senzor na čakanie na dáta
Keďže už vieme, ako sa pripojiť a prijímať dáta z pošty, môžeme teraz napísať senzor, ktorý na ne počká. V mojom prípade nefungovalo hneď napísať operátora, ktorý údaje spracuje, ak nejaké existujú, pretože na základe údajov prijatých z pošty fungujú iné procesy, vrátane tých, ktoré preberajú súvisiace údaje z iných zdrojov (API, telefonovanie , webové metriky atď.) atď.). Uvediem príklad. V CRM systéme sa objavil nový používateľ, o jeho UUID stále nevieme. Potom pri pokuse o príjem dát zo SIP telefónie budeme prijímať hovory viazané na jej UUID, ale nebudeme ich môcť uložiť a správne použiť. V takýchto záležitostiach je dôležité mať na pamäti závislosť údajov, najmä ak sú z rôznych zdrojov. Toto sú, samozrejme, nedostatočné opatrenia na zachovanie integrity údajov, ale v niektorých prípadoch sú nevyhnutné. Áno, a nečinnosť pri obsadzovaní zdrojov je tiež iracionálna.
Náš senzor teda spustí nasledujúce vrcholy grafu, ak sú v pošte čerstvé informácie, a tiež označí predchádzajúce informácie ako irelevantné.
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 TruePrijímame a používame údaje
Na príjem a spracovanie údajov môžete napísať samostatný operátor, môžete použiť hotové. Keďže zatiaľ je logika triviálna - napríklad vziať údaje z listu, navrhujem štandardný 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)
# Стандартное конфигурирование графа
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",
)
# Определяем сенсор
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",
)
# Функция для получения данных из письма
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)
# Описание остальных вершин графа
...
# Задаем связь на графе
mail_check_sensor >> prepare_mail_data
prepare_data >> ...
# Описание остальных потоков управленияMimochodom, ak je vaša firemná pošta aj na mail.ru, nebudete môcť vyhľadávať listy podľa predmetu, odosielateľa atď. Ešte v roku 2016 sľúbili, že ho zavedú, no zrejme si to rozmysleli. Tento problém som vyriešil vytvorením samostatného priečinka pre potrebné písmená a nastavením filtra na potrebné písmená vo webovom rozhraní pošty. Do tohto priečinka sa tak dostanú len potrebné písmená a podmienky na vyhľadávanie, v mojom prípade jednoducho (NEVIDENÉ).
Ak to zhrnieme, máme nasledujúcu postupnosť: skontrolujeme, či existujú nové písmená, ktoré spĺňajú podmienky, ak existujú, potom stiahneme archív pomocou odkazu z posledného písmena.
Pod poslednými bodkami je vynechané, že sa tento archív rozbalí, dáta z archívu sa vyčistia a spracujú a vo výsledku to celé pôjde ďalej do potrubia ETL procesu, ale to už je mimo. rozsah článku. Ak by to bolo zaujímavé a užitočné, tak rád budem pokračovať v popise ETL riešení a ich častí pre Apache Airflow.
Zdroj: hab.com
