ដំណើរការ ETL សម្រាប់ការទទួលបានទិន្នន័យពីអ៊ីមែលនៅក្នុង Apache Airflow

មិនថាបច្ចេកវិទ្យារីកចម្រើនប៉ុណ្ណានោះទេ វិធីសាស្រ្តហួសសម័យតែងតែដើរតាមពីក្រោយការអភិវឌ្ឍន៍។ នេះអាចបណ្តាលមកពីការផ្លាស់ប្តូររលូន កត្តាមនុស្ស តម្រូវការបច្ចេកវិទ្យា ឬអ្វីផ្សេងទៀត។ នៅក្នុងវិស័យដំណើរការទិន្នន័យ ប្រភពទិន្នន័យគឺបង្ហាញឱ្យឃើញច្រើនបំផុតនៅក្នុងផ្នែកនេះ។ មិនថាយើងសុបិនចង់កម្ចាត់វាប៉ុណ្ណានោះទេ ប៉ុន្តែរហូតមកដល់ពេលនេះ ទិន្នន័យមួយផ្នែកត្រូវបានផ្ញើជាសារបន្ទាន់ និងអ៊ីមែល ដោយមិននិយាយអំពីទម្រង់ចាស់ជាងនេះទេ។ ខ្ញុំសូមអញ្ជើញអ្នកឱ្យផ្តាច់ជម្រើសមួយសម្រាប់ Apache Airflow ដោយបង្ហាញពីរបៀបដែលអ្នកអាចយកទិន្នន័យពីអ៊ីមែល។

សម័យបុរេប្រវត្តិសាស្ត្រ

ទិន្នន័យជាច្រើននៅតែត្រូវបានផ្ទេរតាមរយៈអ៊ីមែល ពីទំនាក់ទំនងរវាងបុគ្គលទៅស្តង់ដារនៃអន្តរកម្មរវាងក្រុមហ៊ុន។ វាជាការល្អប្រសិនបើអាចសរសេរចំណុចប្រទាក់ដើម្បីទទួលបានទិន្នន័យ ឬដាក់មនុស្សនៅក្នុងការិយាល័យដែលនឹងបញ្ចូលព័ត៌មាននេះទៅក្នុងប្រភពដែលងាយស្រួលជាង ប៉ុន្តែជារឿយៗវាប្រហែលជាមិនអាចទៅរួចនោះទេ។ ភារកិច្ចជាក់លាក់ដែលខ្ញុំបានប្រឈមគឺការភ្ជាប់ប្រព័ន្ធ CRM ដ៏ល្បីល្បាញទៅនឹងឃ្លាំងទិន្នន័យ ហើយបន្ទាប់មកទៅប្រព័ន្ធ OLAP ។ វាបានកើតឡើងជាប្រវត្តិសាស្ត្រថាសម្រាប់ក្រុមហ៊ុនរបស់យើងការប្រើប្រាស់ប្រព័ន្ធនេះគឺមានភាពងាយស្រួលនៅក្នុងតំបន់ជាក់លាក់នៃអាជីវកម្ម។ ដូច្នេះហើយ អ្នកគ្រប់គ្នាពិតជាចង់ធ្វើប្រតិបត្តិការជាមួយទិន្នន័យពីប្រព័ន្ធភាគីទីបីនេះផងដែរ។ ជាដំបូង ជាការពិត លទ្ធភាពនៃការទទួលបានទិន្នន័យពី API បើកចំហត្រូវបានសិក្សា។ ជាអកុសល API មិនបានគ្របដណ្តប់លើការទទួលបានទិន្នន័យចាំបាច់ទាំងអស់នោះទេ ហើយនិយាយឱ្យចំទៅ វាគឺនៅក្នុងវិធីជាច្រើនដែលមានលក្ខណៈច្របូកច្របល់ ហើយជំនួយផ្នែកបច្ចេកទេសមិនចង់បាន ឬមិនអាចបំពេញបានពាក់កណ្តាលដើម្បីផ្តល់មុខងារទូលំទូលាយបន្ថែមទៀត។ ប៉ុន្តែប្រព័ន្ធនេះបានផ្តល់ឱកាសក្នុងការទទួលទិន្នន័យដែលបាត់តាមប្រៃសណីយ៍តាមកាលកំណត់ក្នុងទម្រង់ជាតំណភ្ជាប់សម្រាប់ការដកឯកសារទុក។

គួរកត់សម្គាល់ថានេះមិនមែនជាករណីតែមួយគត់ដែលអាជីវកម្មចង់ប្រមូលទិន្នន័យពីអ៊ីមែល ឬអ្នកផ្ញើសារបន្ទាន់នោះទេ។ ទោះយ៉ាងណាក៏ដោយ ក្នុងករណីនេះ យើងមិនអាចមានឥទ្ធិពលលើក្រុមហ៊ុនភាគីទីបីដែលផ្តល់ផ្នែកនៃទិន្នន័យតែតាមវិធីនេះទេ។

លំហូរម៉ាស៊ីនត្រជាក់ Apache

ដើម្បីបង្កើតដំណើរការ ETL យើងភាគច្រើនប្រើ Apache Airflow ។ ដើម្បីឱ្យអ្នកអានដែលមិនស្គាល់បច្ចេកវិទ្យានេះយល់កាន់តែច្បាស់អំពីរបៀបដែលវាមើលទៅក្នុងបរិបទ ហើយជាទូទៅ ខ្ញុំនឹងរៀបរាប់អំពីការណែនាំមួយចំនួន។

Apache Airflow គឺជាវេទិកាឥតគិតថ្លៃដែលត្រូវបានប្រើដើម្បីបង្កើត ប្រតិបត្តិ និងត្រួតពិនិត្យដំណើរការ ETL (Extract-Transform-Loading) នៅក្នុង Python ។ គោលគំនិតចម្បងនៅក្នុង Airflow គឺជាក្រាហ្វ acyclic ដឹកនាំ ដែលចំនុចកំពូលនៃក្រាហ្វគឺជាដំណើរការជាក់លាក់ ហើយគែមនៃក្រាហ្វគឺជាលំហូរនៃការគ្រប់គ្រង ឬព័ត៌មាន។ ដំណើរការមួយអាចហៅមុខងារ Python ណាមួយបាន ឬវាអាចមានតក្កវិជ្ជាស្មុគ្រស្មាញពីការហៅមុខងារជាច្រើនជាបន្តបន្ទាប់នៅក្នុងបរិបទនៃថ្នាក់មួយ។ សម្រាប់ប្រតិបត្តិការញឹកញាប់បំផុត មានការអភិវឌ្ឍន៍ដែលត្រៀមរួចជាស្រេចជាច្រើនដែលអាចប្រើបានជាដំណើរការ។ ការអភិវឌ្ឍន៍បែបនេះរួមមានៈ

• ប្រតិបត្តិករ - សម្រាប់ការផ្ទេរទិន្នន័យពីកន្លែងមួយទៅកន្លែងមួយ ឧទាហរណ៍ ពីតារាងទិន្នន័យទៅកាន់ឃ្លាំងទិន្នន័យ។
• ឧបករណ៍ចាប់សញ្ញា - សម្រាប់រង់ចាំការកើតឡើងនៃព្រឹត្តិការណ៍ជាក់លាក់មួយនិងដឹកនាំលំហូរនៃការគ្រប់គ្រងទៅកំពូលជាបន្តបន្ទាប់នៃក្រាហ្វ;
• ទំពក់ - សម្រាប់ប្រតិបត្តិការកម្រិតទាប ឧទាហរណ៍ ដើម្បីទទួលបានទិន្នន័យពីតារាងមូលដ្ឋានទិន្នន័យ (ប្រើក្នុងសេចក្តីថ្លែងការណ៍);
• ហើយដូច្នេះនៅលើ។

វាមិនសមរម្យទេក្នុងការពិពណ៌នាលម្អិតអំពី Apache Airflow នៅក្នុងអត្ថបទនេះ។ ការណែនាំសង្ខេបអាចត្រូវបានមើល នៅទីនេះ ឬ នៅទីនេះ.

ទំពក់សម្រាប់ការទទួលបានទិន្នន័យ

ជាដំបូង ដើម្បីដោះស្រាយបញ្ហា យើងត្រូវសរសេរទំពក់ដែលយើងអាច៖

• ភ្ជាប់ទៅអ៊ីមែល
• ស្វែងរកអក្សរត្រឹមត្រូវ។
• ទទួលទិន្នន័យពីសំបុត្រ។

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, telephony រង្វាស់បណ្តាញ។ល។)។ល។)។ ខ្ញុំនឹងផ្តល់ឱ្យអ្នកនូវឧទាហរណ៍មួយ។ អ្នកប្រើប្រាស់ថ្មីបានបង្ហាញខ្លួននៅក្នុងប្រព័ន្ធ 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