ਅਪਾਚੇ ਏਅਰਫਲੋ ਵਿੱਚ ਈਮੇਲ ਤੋਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ETL ਪ੍ਰਕਿਰਿਆ

ਭਾਵੇਂ ਕਿੰਨੀ ਵੀ ਤਕਨਾਲੋਜੀ ਵਿਕਸਿਤ ਹੋ ਜਾਵੇ, ਪੁਰਾਣੀਆਂ ਪਹੁੰਚਾਂ ਦੀ ਇੱਕ ਸਤਰ ਹਮੇਸ਼ਾ ਵਿਕਾਸ ਦੇ ਪਿੱਛੇ ਰਹਿੰਦੀ ਹੈ। ਇਹ ਇੱਕ ਨਿਰਵਿਘਨ ਪਰਿਵਰਤਨ, ਮਨੁੱਖੀ ਕਾਰਕਾਂ, ਤਕਨੀਕੀ ਲੋੜਾਂ, ਜਾਂ ਕੁਝ ਹੋਰ ਕਾਰਨ ਹੋ ਸਕਦਾ ਹੈ। ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਦੇ ਖੇਤਰ ਵਿੱਚ, ਡੇਟਾ ਸਰੋਤ ਇਸ ਹਿੱਸੇ ਵਿੱਚ ਸਭ ਤੋਂ ਵੱਧ ਪ੍ਰਗਟ ਹੁੰਦੇ ਹਨ. ਭਾਵੇਂ ਅਸੀਂ ਇਸ ਤੋਂ ਛੁਟਕਾਰਾ ਪਾਉਣ ਦਾ ਕਿੰਨਾ ਵੀ ਸੁਪਨਾ ਲੈਂਦੇ ਹਾਂ, ਪਰ ਹੁਣ ਤੱਕ ਡੇਟਾ ਦਾ ਕੁਝ ਹਿੱਸਾ ਤਤਕਾਲ ਸੰਦੇਸ਼ਵਾਹਕਾਂ ਅਤੇ ਈਮੇਲਾਂ ਵਿੱਚ ਭੇਜਿਆ ਜਾਂਦਾ ਹੈ, ਹੋਰ ਪੁਰਾਣੇ ਫਾਰਮੈਟਾਂ ਦਾ ਜ਼ਿਕਰ ਨਹੀਂ ਕਰਨਾ. ਮੈਂ ਤੁਹਾਨੂੰ ਅਪਾਚੇ ਏਅਰਫਲੋ ਲਈ ਵਿਕਲਪਾਂ ਵਿੱਚੋਂ ਇੱਕ ਨੂੰ ਵੱਖ ਕਰਨ ਲਈ ਸੱਦਾ ਦਿੰਦਾ ਹਾਂ, ਇਹ ਦਰਸਾਉਂਦਾ ਹਾਂ ਕਿ ਤੁਸੀਂ ਈਮੇਲਾਂ ਤੋਂ ਡੇਟਾ ਕਿਵੇਂ ਲੈ ਸਕਦੇ ਹੋ।

prehistory

ਬਹੁਤ ਸਾਰਾ ਡੇਟਾ ਅਜੇ ਵੀ ਈਮੇਲ ਦੁਆਰਾ ਪ੍ਰਸਾਰਿਤ ਕੀਤਾ ਜਾਂਦਾ ਹੈ, ਪਰਸਪਰ ਸੰਚਾਰ ਤੋਂ ਕੰਪਨੀਆਂ ਵਿਚਕਾਰ ਆਪਸੀ ਤਾਲਮੇਲ ਦੇ ਮਿਆਰਾਂ ਤੱਕ. ਇਹ ਚੰਗਾ ਹੈ ਜੇਕਰ ਤੁਸੀਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਇੱਕ ਇੰਟਰਫੇਸ ਲਿਖ ਸਕਦੇ ਹੋ ਜਾਂ ਦਫਤਰ ਵਿੱਚ ਉਹਨਾਂ ਲੋਕਾਂ ਨੂੰ ਰੱਖ ਸਕਦੇ ਹੋ ਜੋ ਇਸ ਜਾਣਕਾਰੀ ਨੂੰ ਵਧੇਰੇ ਸੁਵਿਧਾਜਨਕ ਸਰੋਤਾਂ ਵਿੱਚ ਦਾਖਲ ਕਰਨਗੇ, ਪਰ ਅਕਸਰ ਇਹ ਸੰਭਾਵਨਾ ਬਸ ਨਹੀਂ ਹੋ ਸਕਦੀ। ਖਾਸ ਕੰਮ ਜਿਸਦਾ ਮੈਂ ਸਾਹਮਣਾ ਕੀਤਾ ਸੀ ਉਹ ਇੱਕ ਮਸ਼ਹੂਰ CRM ਸਿਸਟਮ ਨੂੰ ਇੱਕ ਡੇਟਾ ਵੇਅਰਹਾਊਸ ਨਾਲ ਜੋੜ ਰਿਹਾ ਸੀ, ਅਤੇ ਫਿਰ ਇੱਕ OLAP ਸਿਸਟਮ ਨਾਲ। ਇਤਿਹਾਸਕ ਤੌਰ 'ਤੇ, ਸਾਡੀ ਕੰਪਨੀ ਲਈ, ਇਸ ਪ੍ਰਣਾਲੀ ਦੀ ਵਰਤੋਂ ਵਪਾਰ ਦੇ ਇੱਕ ਖਾਸ ਖੇਤਰ ਵਿੱਚ ਸੁਵਿਧਾਜਨਕ ਸੀ. ਇਸ ਲਈ, ਹਰ ਕੋਈ ਅਸਲ ਵਿੱਚ ਇਸ ਤੀਜੀ-ਧਿਰ ਪ੍ਰਣਾਲੀ ਦੇ ਡੇਟਾ ਨਾਲ ਵੀ ਕੰਮ ਕਰਨ ਦੇ ਯੋਗ ਹੋਣਾ ਚਾਹੁੰਦਾ ਸੀ. ਸਭ ਤੋਂ ਪਹਿਲਾਂ, ਬੇਸ਼ੱਕ, ਇੱਕ ਓਪਨ API ਤੋਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਦੀ ਸੰਭਾਵਨਾ ਦੀ ਖੋਜ ਕੀਤੀ ਗਈ ਸੀ. ਬਦਕਿਸਮਤੀ ਨਾਲ, API ਨੇ ਸਾਰੇ ਲੋੜੀਂਦੇ ਡੇਟਾ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਕਵਰ ਨਹੀਂ ਕੀਤਾ, ਅਤੇ, ਇਸਨੂੰ ਸਧਾਰਨ ਸ਼ਬਦਾਂ ਵਿੱਚ ਕਹਿਣ ਲਈ, ਇਹ ਵੱਡੇ ਪੱਧਰ 'ਤੇ ਟੇਢੀ ਸੀ, ਅਤੇ ਤਕਨੀਕੀ ਸਹਾਇਤਾ ਹੋਰ ਵਿਆਪਕ ਕਾਰਜਸ਼ੀਲਤਾ ਪ੍ਰਦਾਨ ਕਰਨ ਲਈ ਅੱਧੇ ਰਸਤੇ ਨੂੰ ਪੂਰਾ ਕਰਨ ਵਿੱਚ ਅਸਮਰੱਥ ਸੀ। ਪਰ ਇਸ ਸਿਸਟਮ ਨੇ ਸਮੇਂ-ਸਮੇਂ 'ਤੇ ਪੁਰਾਲੇਖ ਨੂੰ ਡਾਊਨਲੋਡ ਕਰਨ ਲਈ ਲਿੰਕ ਦੇ ਰੂਪ ਵਿੱਚ ਈਮੇਲ ਦੁਆਰਾ ਗੁੰਮ ਹੋਏ ਡੇਟਾ ਨੂੰ ਪ੍ਰਾਪਤ ਕਰਨ ਦਾ ਮੌਕਾ ਪ੍ਰਦਾਨ ਕੀਤਾ।

ਇਹ ਨੋਟ ਕੀਤਾ ਜਾਣਾ ਚਾਹੀਦਾ ਹੈ ਕਿ ਇਹ ਇਕੋ ਇਕ ਮਾਮਲਾ ਨਹੀਂ ਸੀ ਜਿਸ ਵਿਚ ਕਾਰੋਬਾਰ ਈਮੇਲਾਂ ਜਾਂ ਤਤਕਾਲ ਮੈਸੇਂਜਰਾਂ ਤੋਂ ਡਾਟਾ ਇਕੱਠਾ ਕਰਨਾ ਚਾਹੁੰਦਾ ਸੀ. ਹਾਲਾਂਕਿ, ਇਸ ਮਾਮਲੇ ਵਿੱਚ, ਅਸੀਂ ਕਿਸੇ ਤੀਜੀ-ਧਿਰ ਦੀ ਕੰਪਨੀ ਨੂੰ ਪ੍ਰਭਾਵਿਤ ਨਹੀਂ ਕਰ ਸਕਦੇ ਜੋ ਸਿਰਫ ਇਸ ਤਰੀਕੇ ਨਾਲ ਡੇਟਾ ਦਾ ਹਿੱਸਾ ਪ੍ਰਦਾਨ ਕਰਦੀ ਹੈ।

ਅਪਾਚੇ ਏਅਰਫਲੋ

ਈਟੀਐਲ ਪ੍ਰਕਿਰਿਆਵਾਂ ਨੂੰ ਬਣਾਉਣ ਲਈ, ਅਸੀਂ ਅਕਸਰ ਅਪਾਚੇ ਏਅਰਫਲੋ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹਾਂ। ਇੱਕ ਪਾਠਕ ਜੋ ਇਸ ਤਕਨਾਲੋਜੀ ਤੋਂ ਅਣਜਾਣ ਹੈ, ਨੂੰ ਬਿਹਤਰ ਢੰਗ ਨਾਲ ਸਮਝਣ ਲਈ ਕਿ ਇਹ ਸੰਦਰਭ ਵਿੱਚ ਅਤੇ ਆਮ ਤੌਰ 'ਤੇ ਕਿਵੇਂ ਦਿਖਾਈ ਦਿੰਦਾ ਹੈ, ਮੈਂ ਕੁਝ ਸ਼ੁਰੂਆਤੀ ਲੋਕਾਂ ਦਾ ਵਰਣਨ ਕਰਾਂਗਾ।

ਅਪਾਚੇ ਏਅਰਫਲੋ ਇੱਕ ਮੁਫਤ ਪਲੇਟਫਾਰਮ ਹੈ ਜੋ ਪਾਈਥਨ ਵਿੱਚ ETL (ਐਕਸਟ੍ਰੈਕਟ-ਟਰਾਂਸਫਾਰਮ-ਲੋਡਿੰਗ) ਪ੍ਰਕਿਰਿਆਵਾਂ ਨੂੰ ਬਣਾਉਣ, ਚਲਾਉਣ ਅਤੇ ਨਿਗਰਾਨੀ ਕਰਨ ਲਈ ਵਰਤਿਆ ਜਾਂਦਾ ਹੈ। ਏਅਰਫਲੋ ਵਿੱਚ ਮੁੱਖ ਸੰਕਲਪ ਇੱਕ ਨਿਰਦੇਸ਼ਿਤ ਐਸੀਕਲਿਕ ਗ੍ਰਾਫ ਹੈ, ਜਿੱਥੇ ਗ੍ਰਾਫ ਦੇ ਸਿਰਲੇਖ ਖਾਸ ਪ੍ਰਕਿਰਿਆਵਾਂ ਹਨ, ਅਤੇ ਗ੍ਰਾਫ ਦੇ ਕਿਨਾਰੇ ਨਿਯੰਤਰਣ ਜਾਂ ਜਾਣਕਾਰੀ ਦਾ ਪ੍ਰਵਾਹ ਹਨ। ਇੱਕ ਪ੍ਰਕਿਰਿਆ ਕਿਸੇ ਵੀ ਪਾਈਥਨ ਫੰਕਸ਼ਨ ਨੂੰ ਕਾਲ ਕਰ ਸਕਦੀ ਹੈ, ਜਾਂ ਇਸ ਵਿੱਚ ਕਲਾਸ ਦੇ ਸੰਦਰਭ ਵਿੱਚ ਕਈ ਫੰਕਸ਼ਨਾਂ ਨੂੰ ਕ੍ਰਮਵਾਰ ਕਾਲ ਕਰਨ ਦਾ ਵਧੇਰੇ ਗੁੰਝਲਦਾਰ ਤਰਕ ਹੋ ਸਕਦਾ ਹੈ। ਸਭ ਤੋਂ ਆਮ ਓਪਰੇਸ਼ਨਾਂ ਲਈ, ਪਹਿਲਾਂ ਹੀ ਬਹੁਤ ਸਾਰੇ ਤਿਆਰ-ਕੀਤੇ ਵਿਕਾਸ ਹਨ ਜੋ ਪ੍ਰਕਿਰਿਆਵਾਂ ਵਜੋਂ ਵਰਤੇ ਜਾ ਸਕਦੇ ਹਨ। ਅਜਿਹੇ ਵਿਕਾਸ ਵਿੱਚ ਸ਼ਾਮਲ ਹਨ:

• ਓਪਰੇਟਰ - ਡੇਟਾ ਨੂੰ ਇੱਕ ਥਾਂ ਤੋਂ ਦੂਜੀ ਥਾਂ ਤੇ ਟ੍ਰਾਂਸਫਰ ਕਰਨ ਲਈ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ ਡੇਟਾਬੇਸ ਟੇਬਲ ਤੋਂ ਡੇਟਾ ਵੇਅਰਹਾਊਸ ਵਿੱਚ;
• ਸੈਂਸਰ - ਕਿਸੇ ਖਾਸ ਘਟਨਾ ਦੇ ਵਾਪਰਨ ਦੀ ਉਡੀਕ ਕਰਨ ਅਤੇ ਗ੍ਰਾਫ ਦੇ ਅਗਲੇ ਸਿਰਿਆਂ 'ਤੇ ਨਿਯੰਤਰਣ ਦੇ ਪ੍ਰਵਾਹ ਨੂੰ ਨਿਰਦੇਸ਼ਤ ਕਰਨ ਲਈ;
• ਹੁੱਕ - ਹੇਠਲੇ-ਪੱਧਰ ਦੇ ਓਪਰੇਸ਼ਨਾਂ ਲਈ, ਉਦਾਹਰਨ ਲਈ, ਡੇਟਾਬੇਸ ਟੇਬਲ ਤੋਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ (ਕਥਨ ਵਿੱਚ ਵਰਤਿਆ ਜਾਂਦਾ ਹੈ);
• ਅਤੇ ਇਸ ਤਰਾਂ ਹੀ.

ਇਸ ਲੇਖ ਵਿੱਚ ਅਪਾਚੇ ਏਅਰਫਲੋ ਦਾ ਵਿਸਥਾਰ ਵਿੱਚ ਵਰਣਨ ਕਰਨਾ ਅਣਉਚਿਤ ਹੋਵੇਗਾ। ਸੰਖੇਪ ਜਾਣਕਾਰੀ ਦੇਖੀ ਜਾ ਸਕਦੀ ਹੈ ਇੱਥੇ ਜ ਇੱਥੇ.

ਡਾਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਲਈ ਹੁੱਕ

ਸਭ ਤੋਂ ਪਹਿਲਾਂ, ਸਮੱਸਿਆ ਨੂੰ ਹੱਲ ਕਰਨ ਲਈ, ਸਾਨੂੰ ਇੱਕ ਹੁੱਕ ਲਿਖਣ ਦੀ ਲੋੜ ਹੈ ਜਿਸ ਨਾਲ ਅਸੀਂ ਕਰ ਸਕਦੇ ਹਾਂ:

• ਈਮੇਲ ਨਾਲ ਜੁੜੋ
• ਤੁਹਾਨੂੰ ਲੋੜੀਂਦਾ ਪੱਤਰ ਲੱਭੋ;
• ਪੱਤਰ ਤੋਂ ਡੇਟਾ ਪ੍ਰਾਪਤ ਕਰੋ.

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 ਬਾਰੇ ਦੱਸਾਂਗਾ। ਅਪਾਚੇ ਏਅਰਫਲੋ ਕਨੈਕਸ਼ਨ ਪੈਰਾਮੀਟਰਾਂ (ਲੌਗਇਨ, ਪਾਸਵਰਡ, ਪਤਾ, ਅਤੇ ਹੋਰ ਮਾਪਦੰਡ) ਨੂੰ ਸਟੋਰ ਕਰਦਾ ਹੈ ਜਿਨ੍ਹਾਂ ਨੂੰ ਇੱਕ ਸਤਰ ਪਛਾਣਕਰਤਾ ਦੁਆਰਾ ਐਕਸੈਸ ਕੀਤਾ ਜਾ ਸਕਦਾ ਹੈ। ਦ੍ਰਿਸ਼ਟੀਗਤ ਤੌਰ 'ਤੇ, ਕੁਨੈਕਸ਼ਨ ਪ੍ਰਬੰਧਨ ਇਸ ਤਰ੍ਹਾਂ ਦਿਖਾਈ ਦਿੰਦਾ ਹੈ

ਡਾਟਾ ਦੀ ਉਡੀਕ ਕਰਨ ਲਈ ਸੈਂਸਰ

ਕਿਉਂਕਿ ਅਸੀਂ ਪਹਿਲਾਂ ਹੀ ਜਾਣਦੇ ਹਾਂ ਕਿ ਮੇਲ ਤੋਂ ਡੇਟਾ ਕਿਵੇਂ ਕਨੈਕਟ ਕਰਨਾ ਅਤੇ ਪ੍ਰਾਪਤ ਕਰਨਾ ਹੈ, ਅਸੀਂ ਹੁਣ ਉਹਨਾਂ ਦੀ ਉਡੀਕ ਕਰਨ ਲਈ ਇੱਕ ਸੈਂਸਰ ਲਿਖ ਸਕਦੇ ਹਾਂ। ਮੇਰੇ ਕੇਸ ਵਿੱਚ, ਇਸਨੇ ਤੁਰੰਤ ਇੱਕ ਓਪਰੇਟਰ ਲਿਖਣਾ ਕੰਮ ਨਹੀਂ ਕੀਤਾ ਜੋ ਡੇਟਾ ਦੀ ਪ੍ਰਕਿਰਿਆ ਕਰੇਗਾ, ਜੇਕਰ ਕੋਈ ਹੈ, ਕਿਉਂਕਿ ਹੋਰ ਪ੍ਰਕਿਰਿਆਵਾਂ ਮੇਲ ਤੋਂ ਪ੍ਰਾਪਤ ਡੇਟਾ ਦੇ ਅਧਾਰ ਤੇ ਕੰਮ ਕਰਦੀਆਂ ਹਨ, ਜਿਸ ਵਿੱਚ ਉਹ ਵੀ ਸ਼ਾਮਲ ਹਨ ਜੋ ਦੂਜੇ ਸਰੋਤਾਂ (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

ਅਸੀਂ ਡਾਟਾ ਪ੍ਰਾਪਤ ਕਰਦੇ ਹਾਂ ਅਤੇ ਵਰਤਦੇ ਹਾਂ

ਡਾਟਾ ਪ੍ਰਾਪਤ ਕਰਨ ਅਤੇ ਪ੍ਰਕਿਰਿਆ ਕਰਨ ਲਈ, ਤੁਸੀਂ ਇੱਕ ਵੱਖਰਾ ਆਪਰੇਟਰ ਲਿਖ ਸਕਦੇ ਹੋ, ਤੁਸੀਂ ਤਿਆਰ-ਬਣਾਈਆਂ ਦੀ ਵਰਤੋਂ ਕਰ ਸਕਦੇ ਹੋ। ਕਿਉਂਕਿ ਹੁਣ ਤੱਕ ਤਰਕ ਮਾਮੂਲੀ ਹੈ - ਅੱਖਰ ਤੋਂ ਡੇਟਾ ਲੈਣ ਲਈ, ਉਦਾਹਰਨ ਲਈ, ਮੈਂ ਸਟੈਂਡਰਡ ਪਾਈਥਨ ਓਪਰੇਟਰ ਦਾ ਸੁਝਾਅ ਦਿੰਦਾ ਹਾਂ

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 ਹੱਲਾਂ ਅਤੇ ਉਹਨਾਂ ਦੇ ਹਿੱਸਿਆਂ ਦਾ ਵਰਣਨ ਕਰਨਾ ਜਾਰੀ ਰੱਖਾਂਗਾ.

ਸਰੋਤ: www.habr.com