Apache Airflow의 이메일에서 데이터를 가져오기 위한 ETL 프로세스

기술이 아무리 발전해도 시대에 뒤떨어진 접근 방식은 늘 뒤처진다. 이는 순조로운 전환, 인적 요인, 기술적 요구 또는 기타 요인 때문일 수 있습니다. 데이터 처리 분야에서 데이터 소스는 이 부분에서 가장 많이 드러난다. 우리가 이것을 제거하는 것을 아무리 꿈꾸더라도 지금까지 데이터의 일부는 더 오래된 형식은 말할 것도 없고 인스턴트 메신저와 이메일로 전송됩니다. 이메일에서 데이터를 가져올 수 있는 방법을 설명하는 Apache Airflow 옵션 중 하나를 분해하도록 초대합니다.

선사 시대

대인 커뮤니케이션에서 회사 간 상호 작용 표준에 이르기까지 많은 데이터가 여전히 전자 메일을 통해 전송됩니다. 데이터를 얻기 위한 인터페이스를 작성하거나 이 정보를 더 편리한 소스에 입력할 사람들을 사무실에 배치할 수 있다면 좋지만 종종 이것은 단순히 불가능할 수도 있습니다. 내가 직면한 구체적인 작업은 악명 높은 CRM 시스템을 데이터 웨어하우스에 연결한 다음 OLAP 시스템에 연결하는 것이었습니다. 역사적으로 우리 회사의 경우 특정 비즈니스 영역에서이 시스템을 사용하는 것이 편리했습니다. 따라서 모두가 이 타사 시스템의 데이터로도 작동할 수 있기를 원했습니다. 물론 먼저 공개 API에서 데이터를 얻을 수 있는 가능성에 대해 연구했습니다. 불행히도 API는 필요한 모든 데이터를 얻는 것을 다루지 않았고, 간단히 말해서 여러 면에서 비뚤어져 있었고 기술 지원은 더 포괄적인 기능을 제공하기를 원하지 않았거나 만날 수 없었습니다. 그러나이 시스템은 누락 된 데이터를 아카이브 언로드 링크 형태로 메일로 주기적으로받을 수있는 기회를 제공했습니다.

비즈니스가 이메일이나 인스턴트 메신저에서 데이터를 수집하기를 원했던 경우는 이것이 유일한 경우가 아니라는 점에 유의해야 합니다. 그러나 이 경우 이러한 방식으로만 데이터의 일부를 제공하는 제XNUMX자 회사에 영향을 미칠 수 없습니다.

아파치 에어 플로우

ETL 프로세스를 구축하기 위해 Apache Airflow를 가장 자주 사용합니다. 이 기술에 익숙하지 않은 독자가 상황과 일반적으로 어떻게 보이는지 더 잘 이해할 수 있도록 몇 가지 소개를 설명하겠습니다.

Apache Airflow는 Python에서 ETL(Extract-Transform-Loading) 프로세스를 빌드, 실행 및 모니터링하는 데 사용되는 무료 플랫폼입니다. Airflow의 주요 개념은 그래프의 정점이 특정 프로세스이고 그래프의 가장자리가 제어 또는 정보의 흐름인 방향성 비순환 그래프입니다. 프로세스는 단순히 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, 전화 통신)에서 관련 데이터를 가져오는 프로세스를 포함하여 메일에서 받은 데이터를 기반으로 다른 프로세스가 작동하기 때문에 데이터를 처리할 연산자를 즉시 ​​작성하지 못했습니다. , 웹 메트릭 등) 등). 예를 들어 보겠습니다. 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용 부품에 대해 기꺼이 계속 설명하겠습니다.

출처 : habr.com