תהליך ETL לקבלת נתונים מאימייל ב- Apache Airflow

לא משנה כמה הטכנולוגיה מתפתחת, שורה של גישות מיושנות תמיד נגררת אחרי הפיתוח. זה עשוי לנבוע ממעבר חלק, גורמים אנושיים, צרכים טכנולוגיים או משהו אחר. בתחום עיבוד הנתונים, מקורות הנתונים הם החושפניים ביותר בחלק זה. לא משנה כמה אנחנו חולמים להיפטר מזה, אבל עד כה חלק מהנתונים נשלחים בהודעות מיידיות ובמיילים, שלא לדבר על פורמטים ארכאיים יותר. אני מזמין אותך לפרק את אחת האפשרויות עבור Apache Airflow, וממחיש כיצד ניתן לקחת נתונים ממיילים.

פרהיסטוריה

מידע רב עדיין מועבר באמצעות דואר אלקטרוני, מתקשורת בין אישית ועד לסטנדרטים של אינטראקציה בין חברות. זה טוב אם אפשר לכתוב ממשק להשגת נתונים או להכניס אנשים למשרד שיכניסו את המידע הזה למקורות נוחים יותר, אבל לעתים קרובות זה פשוט לא אפשרי. המשימה הספציפית שעמדתי בפני הייתה חיבור מערכת ה-CRM הידועה לשמצה למחסן הנתונים, ולאחר מכן למערכת OLAP. כך קרה באופן היסטורי שעבור החברה שלנו השימוש במערכת זו היה נוח בתחום מסוים של עסקים. לכן, כולם באמת רצו להיות מסוגלים לפעול גם עם נתונים ממערכת צד שלישי זו. קודם כל, כמובן, נבדקה האפשרות לקבל נתונים מ-API פתוח. לרוע המזל, ה-API לא כיסה את קבלת כל הנתונים הדרושים, ובמונחים פשוטים, הוא היה עקום במובנים רבים, והתמיכה הטכנית לא רצתה או לא יכלה להיפגש באמצע הדרך כדי לספק פונקציונליות מקיפה יותר. אבל מערכת זו סיפקה את ההזדמנות לקבל מעת לעת את הנתונים החסרים בדואר בצורה של קישור לפריקת הארכיון.

יש לציין שזה לא היה המקרה היחיד בו רצה העסק לאסוף נתונים ממיילים או מסרים מידיים. עם זאת, במקרה זה, לא יכולנו להשפיע על חברת צד שלישי המספקת חלק מהנתונים רק בדרך זו.

זרימת אוויר של אפאצ'י

כדי לבנות תהליכי ETL, אנו משתמשים לרוב ב- Apache Airflow. כדי שקורא שלא מכיר את הטכנולוגיה הזו יבין טוב יותר איך היא נראית בהקשר ובכלל, אתאר כמה מבוא.

Apache Airflow היא פלטפורמה חינמית המשמשת לבנייה, ביצוע וניטור תהליכי ETL (Extract-Transform-Loading) ב-Python. הרעיון המרכזי ב- Airflow הוא גרף א-ציקלי מכוון, כאשר קודקודי הגרף הם תהליכים ספציפיים, וקצוות הגרף הם זרימת השליטה או המידע. תהליך יכול פשוט לקרוא לכל פונקציה של Python, או שיכול להיות לו היגיון מורכב יותר מקריאת מספר פונקציות ברצף בהקשר של מחלקה. עבור הפעולות השכיחות ביותר, יש כבר הרבה פיתוחים מוכנים שיכולים לשמש כתהליכים. פיתוחים כאלה כוללים:

• מפעילים - להעברת נתונים ממקום למקום, למשל מטבלת מסד נתונים למחסן נתונים;
• חיישנים - להמתנה להתרחשות של אירוע מסוים והכוונת זרימת השליטה לקודקודים הבאים של הגרף;
• hooks - עבור פעולות ברמה נמוכה יותר, למשל, כדי לקבל נתונים מטבלת מסד נתונים (בשימוש בהצהרות);
• וכו '

זה יהיה לא הולם לתאר את זרימת האוויר של Apache בפירוט במאמר זה. ניתן לצפות במבואות קצרים כאן או כאן.

וו לקבלת נתונים

קודם כל, כדי לפתור את הבעיה, אנחנו צריכים לכתוב וו שאיתו נוכל:

• להתחבר למייל
• למצוא את האות הנכונה
• לקבל נתונים מהמכתב.

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_id

ההיגיון הוא כזה: אנחנו מתחברים, מוצאים את האות האחרונה הכי רלוונטית, אם יש עוד, אנחנו מתעלמים מהם. פונקציה זו משמשת, כי אותיות מאוחרות יותר מכילות את כל הנתונים של קודמות. אם זה לא המקרה, אז אתה יכול להחזיר מערך של כל האותיות, או לעבד את הראשונה, ואת השאר במעבר הבא. באופן כללי, הכל, כמו תמיד, תלוי במשימה.

אנו מוסיפים שתי פונקציות עזר להוק: להורדת קובץ ולהורדת קובץ באמצעות קישור ממייל. אגב, ניתן להעביר אותם למפעיל, זה תלוי בתדירות השימוש בפונקציונליות זו. מה עוד להוסיף לקרס, שוב, תלוי במשימה: אם קבצים מתקבלים מיד במכתב, אז אתה יכול להוריד קבצים מצורפים למכתב, אם הנתונים מתקבלים במכתב, אז אתה צריך לנתח את המכתב, וכו ' במקרה שלי, המכתב מגיע עם קישור אחד לארכיון, אותו אני צריך לשים במקום מסוים ולהתחיל בתהליך העיבוד.

    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)

הקוד פשוט, כך שהוא כמעט לא צריך הסבר נוסף. אני רק אספר לך על קו הקסם imap_conn_id. Apache Airflow מאחסן פרמטרים של חיבור (כניסה, סיסמה, כתובת ופרמטרים אחרים) שניתן לגשת אליהם באמצעות מזהה מחרוזת. מבחינה ויזואלית, ניהול החיבורים נראה כך

חיישן להמתנה לנתונים

מכיוון שאנו כבר יודעים כיצד להתחבר ולקבל נתונים מהדואר, אנו יכולים כעת לכתוב חיישן שיחכה להם. במקרה שלי, זה לא עבד לכתוב מיד מפעיל שיעבד את הנתונים, אם יש, כי תהליכים אחרים עובדים על סמך הנתונים המתקבלים מהדואר, כולל אלה שלוקחים נתונים קשורים ממקורות אחרים (API, טלפוניה , מדדי אינטרנט וכו'). וכו'). אני אתן לך דוגמה. משתמש חדש הופיע במערכת ה-CRM, ועדיין איננו יודעים על ה-UUID שלו. לאחר מכן, כאשר מנסים לקבל נתונים מטלפוניה SIP, נקבל שיחות הקשורות ל-UUID שלה, אך לא נוכל לשמור ולהשתמש בהן בצורה נכונה. בנושאים כאלה, חשוב לזכור את התלות של הנתונים, במיוחד אם הם ממקורות שונים. אלו, כמובן, אמצעים לא מספיקים לשמירה על שלמות הנתונים, אך במקרים מסוימים הם נחוצים. כן, והבטלה כדי לכבוש משאבים היא גם לא הגיונית.

לפיכך, החיישן שלנו ישיק קודקודים של הגרף אם יש מידע טרי בדואר, וגם יסמן את המידע הקודם כלא רלוונטי.

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 True

אנו מקבלים נתונים ומשתמשים בהם

כדי לקבל ולעבד נתונים, אתה יכול לכתוב אופרטור נפרד, אתה יכול להשתמש באלה מוכנים. מכיוון שעד כה ההיגיון הוא טריוויאלי - לקחת נתונים מהמכתב, למשל, אני מציע את ה-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": ["[email protected]"],
    "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 >> ...
# Описание остальных потоков управления

אגב, אם הדואר הארגוני שלך נמצא גם ב-mail.ru, אז לא תוכל לחפש מכתבים לפי נושא, שולח וכו'. עוד ב-2016, הם הבטיחו להציג את זה, אבל כנראה שינו את דעתם. פתרתי בעיה זו על ידי יצירת תיקייה נפרדת לאותיות הנחוצות והגדרת מסנן לאותיות הדרושות בממשק האינטרנט של הדואר. לפיכך, רק המכתבים והתנאים הדרושים לחיפוש, במקרה שלי, פשוט (UNSEEN) נכנסים לתיקיה זו.

לסיכום, יש לנו את הרצף הבא: אנחנו בודקים אם יש אותיות חדשות שעומדות בתנאים, אם יש, אז אנחנו מורידים את הארכיון באמצעות הקישור מהמכתב האחרון.
מתחת לנקודות האחרונות, נשמט שהארכיון הזה יפורק, הנתונים מהארכיון ינוקו ויעובדו, וכתוצאה מכך, כל העניין ילך רחוק יותר לצינור של תהליך ETL, אבל זה כבר מעבר היקף המאמר. אם זה יצא מעניין ושימושי, אז בשמחה אמשיך לתאר פתרונות ETL והחלקים שלהם עבור Apache Airflow.

מקור: www.habr.com