ETL პროცესი ელფოსტიდან მონაცემების მისაღებად Apache Airflow-ში

რაც არ უნდა განვითარდეს ტექნოლოგია, მოძველებული მიდგომების სერია ყოველთვის ჩამორჩება განვითარების უკან. ეს შეიძლება გამოწვეული იყოს გლუვი გადასვლით, ადამიანური ფაქტორებით, ტექნოლოგიური საჭიროებებით ან სხვა რამით. მონაცემთა დამუშავების სფეროში, ამ ნაწილში ყველაზე გამოვლენილია მონაცემთა წყაროები. რაც არ უნდა ვოცნებობდეთ ამის თავიდან აცილებაზე, მაგრამ ჯერჯერობით მონაცემების ნაწილი იგზავნება მყისიერ მესინჯერებსა და ელ.წერილებში, რომ აღარაფერი ვთქვათ უფრო არქაულ ფორმატებში. მე გეპატიჟებით დაშალოთ Apache Airflow-ის ერთ-ერთი ვარიანტი, რომელიც ასახავს, ​​თუ როგორ შეგიძლიათ მიიღოთ მონაცემები ელფოსტიდან.

წინაისტორია

ბევრი მონაცემი კვლავ გადადის ელექტრონული ფოსტით, ინტერპერსონალური კომუნიკაციებიდან კომპანიებს შორის ურთიერთქმედების სტანდარტებამდე. კარგია, თუ შესაძლებელია მონაცემთა მოსაპოვებლად ინტერფეისის დაწერა ან ოფისში ადამიანების განთავსება, რომლებიც ამ ინფორმაციას უფრო მოსახერხებელ წყაროებში შეიტანენ, მაგრამ ხშირად ეს შეიძლება უბრალოდ შეუძლებელი იყოს. კონკრეტული ამოცანა, რომელიც მე წავაწყდი, იყო ცნობილი CRM სისტემის დაკავშირება მონაცემთა საწყობთან, შემდეგ კი OLAP სისტემასთან. ასე მოხდა ისტორიულად, რომ ჩვენი კომპანიისთვის ამ სისტემის გამოყენება მოსახერხებელი იყო ბიზნესის კონკრეტულ სფეროში. ამიტომ, ყველას ძალიან სურდა ამ მესამე მხარის სისტემის მონაცემებით მუშაობაც. უპირველეს ყოვლისა, რა თქმა უნდა, შესწავლილი იქნა ღია API-დან მონაცემების მოპოვების შესაძლებლობა. სამწუხაროდ, API არ მოიცავდა ყველა საჭირო მონაცემის მიღებას და, მარტივი სიტყვებით რომ ვთქვათ, ის მრავალი თვალსაზრისით მცდარი იყო და ტექნიკური მხარდაჭერა არ სურდა ან ვერ ხვდებოდა შუა გზაზე უფრო სრულყოფილი ფუნქციონირების უზრუნველსაყოფად. მაგრამ ეს სისტემა საშუალებას აძლევდა პერიოდულად მიეღო დაკარგული მონაცემები ფოსტით არქივის გადმოტვირთვის ბმულის სახით.

უნდა აღინიშნოს, რომ ეს არ იყო ერთადერთი შემთხვევა, როდესაც ბიზნესს სურდა მონაცემების შეგროვება ელ.წერილებიდან ან მყისიერი მესინჯერებიდან. თუმცა, ამ შემთხვევაში, ჩვენ ვერ მოვახერხეთ გავლენა მესამე მხარის კომპანიაზე, რომელიც მხოლოდ ამ გზით აწვდის მონაცემთა ნაწილს.

აპაჩის ჰაერის ნაკადი

ETL პროცესების შესაქმნელად, ჩვენ ყველაზე ხშირად ვიყენებთ Apache Airflow-ს. იმისათვის, რომ მკითხველმა, რომელიც არ იცნობს ამ ტექნოლოგიას, უკეთ გაიგოს, როგორ გამოიყურება ის კონტექსტში და ზოგადად, მე აღვწერ რამდენიმე შესავალს.

Apache Airflow არის უფასო პლატფორმა, რომელიც გამოიყენება Python-ში ETL (Extract-Transform-Loading) პროცესების შესაქმნელად, შესასრულებლად და მონიტორინგისთვის. Airflow-ის მთავარი კონცეფცია არის მიმართული აციკლური გრაფიკი, სადაც გრაფის წვეროები არის სპეციფიკური პროცესები, ხოლო გრაფის კიდეები არის კონტროლის ან ინფორმაციის ნაკადი. პროცესს შეუძლია უბრალოდ გამოიძახოს პითონის ნებისმიერი ფუნქცია, ან შეიძლება ჰქონდეს უფრო რთული ლოგიკა კლასის კონტექსტში რამდენიმე ფუნქციის თანმიმდევრული გამოძახებისგან. ყველაზე ხშირი ოპერაციებისთვის, უკვე არსებობს მრავალი მზა განვითარება, რომელიც შეიძლება გამოყენებულ იქნას როგორც პროცესები. ასეთი მოვლენები მოიცავს:

• ოპერატორები - მონაცემების ერთი ადგილიდან მეორეზე გადასატანად, მაგალითად, მონაცემთა ბაზის ცხრილიდან მონაცემთა საწყობში;
• სენსორები - გარკვეული მოვლენის მოლოდინში და კონტროლის ნაკადის მიმართულება გრაფის შემდგომ წვეროებზე;
• hooks - ქვედა დონის ოპერაციებისთვის, მაგალითად, მონაცემთა ბაზის ცხრილიდან მონაცემების მისაღებად (გამოიყენება განცხადებებში);
• ა.შ.

ამ სტატიაში Apache Airflow დეტალურად აღწერა არ იქნება მიზანშეწონილი. მოკლე შესავალი შეგიძლიათ იხილოთ აქ ან აქ.

Hook მონაცემების მისაღებად

უპირველეს ყოვლისა, პრობლემის გადასაჭრელად, ჩვენ უნდა დავწეროთ კაკალი, რომლითაც შეგვიძლია:

• დაუკავშირდით ელ.წერილს
• იპოვნეთ სწორი ასო
• მიიღეთ მონაცემები წერილიდან.

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 წელს დაჰპირდნენ მის დანერგვას, მაგრამ, როგორც ჩანს, გადაიფიქრეს. მე გადავწყვიტე ეს პრობლემა საჭირო ასოებისთვის ცალკე საქაღალდის შექმნით და ფოსტის ვებ ინტერფეისში საჭირო ასოებისთვის ფილტრის დაყენებით. ამრიგად, ამ საქაღალდეში მხოლოდ ძიების საჭირო ასოები და პირობები, ჩემს შემთხვევაში, უბრალოდ (გაურკვეველია).

შეჯამებით, გვაქვს შემდეგი თანმიმდევრობა: ვამოწმებთ არის თუ არა ახალი ასოები, რომლებიც აკმაყოფილებენ პირობებს, თუ არის, შემდეგ ჩამოვტვირთავთ არქივს ბოლო ასოდან ბმულის გამოყენებით.
ბოლო წერტილების ქვეშ გამოტოვებულია, რომ ეს არქივი გაიხსნება, არქივიდან მონაცემები გასუფთავდება და დამუშავდება და შედეგად, ყველაფერი უფრო შორს წავა ETL პროცესის მილსადენამდე, მაგრამ ეს უკვე სცილდება. სტატიის ფარგლები. თუ საინტერესო და სასარგებლო აღმოჩნდა, მაშინ მე სიამოვნებით გავაგრძელებ ETL გადაწყვეტილებების და მათი ნაწილების აღწერას Apache Airflow-ისთვის.

წყარო: www.habr.com