ຂະບວນການ ETL ສໍາລັບການໄດ້ຮັບຂໍ້ມູນຈາກອີເມລ໌ໃນ Apache Airflow

ບໍ່ວ່າເທກໂນໂລຍີຈະພັດທະນາຫຼາຍປານໃດ, ວິທີການທີ່ລ້າສະໄຫມມັກຈະຕິດຕາມການພັດທະນາ. ນີ້ອາດຈະເປັນຍ້ອນການຫັນປ່ຽນທີ່ລຽບງ່າຍ, ປັດໃຈຂອງມະນຸດ, ຄວາມຕ້ອງການດ້ານເຕັກໂນໂລຢີ, ຫຼືສິ່ງອື່ນ. ໃນຂົງເຂດການປຸງແຕ່ງຂໍ້ມູນ, ແຫຼ່ງຂໍ້ມູນແມ່ນເປີດເຜີຍຫຼາຍທີ່ສຸດໃນສ່ວນນີ້. ບໍ່ວ່າພວກເຮົາຝັນຢາກກໍາຈັດສິ່ງນີ້ຫຼາຍປານໃດ, ແຕ່ມາຮອດປັດຈຸບັນຂໍ້ມູນສ່ວນຫນຶ່ງແມ່ນຖືກສົ່ງໄປໃນຕົວສົ່ງຂໍ້ຄວາມແລະອີເມວ, ບໍ່ໄດ້ກ່າວເຖິງຮູບແບບເກົ່າແກ່ຫຼາຍ. ຂ້າພະເຈົ້າຂໍເຊື້ອເຊີນທ່ານໃຫ້ຖອດຫນຶ່ງໃນທາງເລືອກສໍາລັບ Apache Airflow, ສະແດງວິທີທີ່ທ່ານສາມາດເອົາຂໍ້ມູນຈາກອີເມວ.

ປະວັດສາດ

ຂໍ້ມູນຈໍານວນຫລາຍຍັງຖືກໂອນຜ່ານທາງອີເມລ໌, ຈາກການສື່ສານລະຫວ່າງບຸກຄົນກັບມາດຕະຖານການພົວພັນລະຫວ່າງບໍລິສັດ. ມັນເປັນການດີຖ້າມັນເປັນໄປໄດ້ທີ່ຈະຂຽນການໂຕ້ຕອບເພື່ອໃຫ້ໄດ້ຂໍ້ມູນຫຼືເອົາຄົນຢູ່ໃນຫ້ອງການທີ່ຈະເອົາຂໍ້ມູນນີ້ເຂົ້າໄປໃນແຫຼ່ງທີ່ສະດວກກວ່າ, ແຕ່ເລື້ອຍໆນີ້ອາດຈະບໍ່ເປັນໄປໄດ້. ວຽກງານສະເພາະທີ່ຂ້າພະເຈົ້າປະເຊີນແມ່ນການເຊື່ອມຕໍ່ລະບົບ CRM ທີ່ມີຊື່ສຽງກັບຄັງຂໍ້ມູນ, ແລະຫຼັງຈາກນັ້ນກັບລະບົບ OLAP. ມັນເກີດຂຶ້ນໃນປະຫວັດສາດວ່າສໍາລັບບໍລິສັດຂອງພວກເຮົາການນໍາໃຊ້ລະບົບນີ້ແມ່ນສະດວກໃນພື້ນທີ່ທຸລະກິດໂດຍສະເພາະ. ເພາະສະນັ້ນ, ທຸກຄົນກໍ່ຕ້ອງການສາມາດດໍາເນີນການກັບຂໍ້ມູນຈາກລະບົບພາກສ່ວນທີສາມນີ້ເຊັ່ນດຽວກັນ. ກ່ອນອື່ນ ໝົດ, ແນ່ນອນ, ຄວາມເປັນໄປໄດ້ຂອງການໄດ້ຮັບຂໍ້ມູນຈາກ API ເປີດໄດ້ຖືກສຶກສາ. ແຕ່ຫນ້າເສຍດາຍ, API ບໍ່ໄດ້ກວມເອົາການໄດ້ຮັບຂໍ້ມູນທີ່ຈໍາເປັນທັງຫມົດ, ແລະ, ໃນຄໍາສັບທີ່ງ່າຍດາຍ, ມັນຢູ່ໃນຫຼາຍທາງທີ່ຫຍາບຄາຍ, ແລະການສະຫນັບສະຫນູນດ້ານວິຊາການບໍ່ຕ້ອງການຫຼືບໍ່ສາມາດຕອບສະຫນອງເຄິ່ງຫນຶ່ງເພື່ອໃຫ້ມີຫນ້າທີ່ທີ່ສົມບູນແບບຫຼາຍຂຶ້ນ. ແຕ່ລະບົບນີ້ໃຫ້ໂອກາດທີ່ຈະໄດ້ຮັບຂໍ້ມູນທີ່ຂາດຫາຍໄປເປັນໄລຍະທາງໄປສະນີໃນຮູບແບບຂອງການເຊື່ອມຕໍ່ສໍາລັບການ unloading ຮວບຮວມ.

ມັນຄວນຈະສັງເກດວ່ານີ້ບໍ່ແມ່ນກໍລະນີດຽວທີ່ທຸລະກິດຕ້ອງການເກັບກໍາຂໍ້ມູນຈາກອີເມລ໌ຫຼືຜູ້ສົ່ງຂໍ້ຄວາມທັນທີ. ຢ່າງໃດກໍຕາມ, ໃນກໍລະນີນີ້, ພວກເຮົາບໍ່ສາມາດມີອິດທິພົນຕໍ່ບໍລິສັດພາກສ່ວນທີສາມທີ່ສະຫນອງສ່ວນຫນຶ່ງຂອງຂໍ້ມູນພຽງແຕ່ໃນວິທີການນີ້.

apache airflow

ເພື່ອສ້າງຂະບວນການ ETL, ພວກເຮົາສ່ວນຫຼາຍແມ່ນໃຊ້ Apache Airflow. ເພື່ອໃຫ້ຜູ້ອ່ານທີ່ບໍ່ຄຸ້ນເຄີຍກັບເທກໂນໂລຍີນີ້ເຂົ້າໃຈດີກວ່າວ່າມັນມີລັກສະນະແນວໃດໃນສະພາບການແລະໂດຍທົ່ວໄປ, ຂ້ອຍຈະອະທິບາຍສອງສາມຂໍ້ແນະນໍາ.

Apache Airflow ເປັນແພລະຕະຟອມຟຣີທີ່ຖືກນໍາໃຊ້ເພື່ອສ້າງ, ປະຕິບັດແລະຕິດຕາມຂະບວນການ ETL (Extract-Transform-Loading) ໃນ Python. ແນວຄວາມຄິດຕົ້ນຕໍໃນ Airflow ແມ່ນກາຟ acyclic ຊີ້, ບ່ອນທີ່ຈຸດຂອງເສັ້ນສະແດງແມ່ນຂະບວນການສະເພາະ, ແລະຂອບຂອງກາຟແມ່ນການໄຫຼເຂົ້າຂອງການຄວບຄຸມຫຼືຂໍ້ມູນ. ຂະບວນການສາມາດເອີ້ນຟັງຊັນ Python ໃດໆ, ຫຼືມັນສາມາດມີເຫດຜົນທີ່ສັບສົນຫຼາຍຈາກການເອີ້ນຫຼາຍຫນ້າທີ່ຕາມລໍາດັບໃນສະພາບການຂອງຫ້ອງຮຽນ. ສໍາລັບການປະຕິບັດງານເລື້ອຍໆທີ່ສຸດ, ມີການພັດທະນາທີ່ກຽມພ້ອມຫຼາຍແລ້ວທີ່ສາມາດຖືກນໍາໃຊ້ເປັນຂະບວນການ. ການພັດທະນາດັ່ງກ່າວປະກອບມີ:

• operators - ສໍາລັບການໂອນຂໍ້ມູນຈາກບ່ອນຫນຶ່ງໄປຫາບ່ອນອື່ນ, ຕົວຢ່າງ, ຈາກຕາຕະລາງຖານຂໍ້ມູນໄປຫາຄັງຂໍ້ມູນ;
• ເຊັນເຊີ - ສໍາລັບການລໍຖ້າການປະກົດຕົວຂອງເຫດການສະເພາະໃດຫນຶ່ງແລະຊີ້ນໍາການໄຫຼເຂົ້າຂອງການຄວບຄຸມໄປສູ່ຈຸດໆຕໍ່ມາຂອງກາຟ;
• hooks - ສໍາລັບການປະຕິບັດລະດັບຕ່ໍາ, ສໍາລັບການຍົກຕົວຢ່າງ, ເພື່ອໃຫ້ໄດ້ຮັບຂໍ້ມູນຈາກຕາຕະລາງຖານຂໍ້ມູນ (ໃຊ້ໃນຄໍາຖະແຫຼງການ);
• ແລະອື່ນໆ.

ມັນຈະບໍ່ເຫມາະສົມທີ່ຈະອະທິບາຍ Apache Airflow ໃນລາຍລະອຽດໃນບົດຄວາມນີ້. ການແນະນໍາສັ້ນໆສາມາດເບິ່ງໄດ້ ທີ່ນີ້ ຫຼື ທີ່ນີ້.

Hook ສໍາລັບການຮັບຂໍ້ມູນ

ກ່ອນອື່ນ ໝົດ, ເພື່ອແກ້ໄຂບັນຫາ, ພວກເຮົາຕ້ອງຂຽນ 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

ເຫດຜົນແມ່ນນີ້: ພວກເຮົາເຊື່ອມຕໍ່, ຊອກຫາຕົວອັກສອນທີ່ກ່ຽວຂ້ອງທີ່ສຸດ, ຖ້າມີຄົນອື່ນ, ພວກເຮົາບໍ່ສົນໃຈພວກມັນ. ຟັງຊັນນີ້ຖືກນໍາໃຊ້, ເພາະວ່າຕົວອັກສອນຕໍ່ມາມີຂໍ້ມູນທັງຫມົດຂອງຕົວຫນັງສືກ່ອນຫນ້າ. ຖ້ານີ້ບໍ່ແມ່ນກໍລະນີ, ທ່ານສາມາດສົ່ງຄືນ array ຂອງຕົວອັກສອນທັງຫມົດ, ຫຼືປຸງແຕ່ງຕົວອັກສອນທໍາອິດ, ແລະສ່ວນທີ່ເຫຼືອໃນ pass ຕໍ່ໄປ. ໂດຍທົ່ວໄປ, ທຸກສິ່ງທຸກຢ່າງ, ຕາມສະເຫມີ, ແມ່ນຂຶ້ນກັບວຽກງານ.

ພວກເຮົາເພີ່ມສອງຫນ້າທີ່ຊ່ວຍສໍາລັບ hook: ສໍາລັບການດາວໂຫຼດໄຟລ໌ແລະສໍາລັບການດາວໂຫຼດໄຟລ໌ໂດຍໃຊ້ການເຊື່ອມຕໍ່ຈາກອີເມລ໌. ໂດຍວິທີທາງການ, ພວກເຂົາສາມາດຖືກຍ້າຍໄປຍັງຜູ້ປະຕິບັດງານ, ມັນຂຶ້ນກັບຄວາມຖີ່ຂອງການນໍາໃຊ້ຫນ້າທີ່ນີ້. ສິ່ງອື່ນທີ່ຈະເພີ່ມໃສ່ hook, ອີກເທື່ອຫນຶ່ງ, ແມ່ນຂຶ້ນກັບວຽກງານ: ຖ້າໄຟລ໌ໄດ້ຮັບທັນທີໃນຈົດຫມາຍ, ຫຼັງຈາກນັ້ນທ່ານສາມາດດາວໂຫລດໄຟລ໌ແນບກັບຈົດຫມາຍ, ຖ້າຂໍ້ມູນໄດ້ຮັບໃນຈົດຫມາຍ, ຫຼັງຈາກນັ້ນທ່ານຈໍາເປັນຕ້ອງແຍກຈົດຫມາຍ, ແລະອື່ນໆ ໃນກໍລະນີຂອງຂ້ອຍ, ຈົດຫມາຍມາພ້ອມກັບການເຊື່ອມຕໍ່ຫນຶ່ງໄປຫາບ່ອນເກັບມ້ຽນ, ເຊິ່ງຂ້ອຍຈໍາເປັນຕ້ອງໃສ່ຢູ່ໃນສະຖານທີ່ທີ່ແນ່ນອນແລະເລີ່ມຕົ້ນຂະບວນການປຸງແຕ່ງຕື່ມອີກ.

    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)

ລະຫັດແມ່ນງ່າຍດາຍ, ສະນັ້ນມັນບໍ່ຄ່ອຍຕ້ອງການຄໍາອະທິບາຍຕື່ມອີກ. ຂ້າພະເຈົ້າພຽງແຕ່ຈະບອກທ່ານກ່ຽວກັບເສັ້ນ magic imap_conn_id. Apache Airflow ເກັບຮັກສາຕົວກໍານົດການເຊື່ອມຕໍ່ (ເຂົ້າສູ່ລະບົບ, ລະຫັດຜ່ານ, ທີ່ຢູ່, ແລະຕົວກໍານົດການອື່ນໆ) ທີ່ສາມາດເຂົ້າເຖິງໄດ້ໂດຍຕົວລະບຸສະຕຣິງ. ສາຍຕາ, ການຈັດການການເຊື່ອມຕໍ່ເບິ່ງຄືວ່ານີ້

ເຊັນເຊີລໍຖ້າຂໍ້ມູນ

ເນື່ອງຈາກພວກເຮົາຮູ້ວິທີເຊື່ອມຕໍ່ ແລະຮັບຂໍ້ມູນຈາກເມລແລ້ວ, ຕອນນີ້ພວກເຮົາສາມາດຂຽນເຊັນເຊີເພື່ອລໍຖ້າພວກມັນໄດ້. ໃນກໍລະນີຂອງຂ້ອຍ, ມັນບໍ່ໄດ້ເຮັດວຽກໃນການຂຽນຕົວປະຕິບັດການທັນທີທີ່ຈະປະມວນຜົນຂໍ້ມູນ, ຖ້າມີ, ເພາະວ່າຂະບວນການອື່ນໆເຮັດວຽກໂດຍອີງໃສ່ຂໍ້ມູນທີ່ໄດ້ຮັບຈາກເມລ, ລວມທັງຂໍ້ມູນທີ່ເອົາຂໍ້ມູນທີ່ກ່ຽວຂ້ອງຈາກແຫຼ່ງອື່ນໆ (API, ໂທລະສັບ. , web metrics, etc.) ແລະອື່ນໆ). ຂ້ອຍຈະຍົກຕົວຢ່າງໃຫ້ເຈົ້າ. ຜູ້ໃຊ້ໃຫມ່ໄດ້ປາກົດຢູ່ໃນລະບົບ CRM, ແລະພວກເຮົາຍັງບໍ່ຮູ້ກ່ຽວກັບ UUID ຂອງລາວ. ຈາກນັ້ນ, ເມື່ອພະຍາຍາມຮັບຂໍ້ມູນຈາກໂທລະສັບ SIP, ພວກເຮົາຈະຮັບສາຍທີ່ເຊື່ອມຕໍ່ກັບ UUID ຂອງມັນ, ແຕ່ພວກເຮົາບໍ່ສາມາດບັນທຶກ ແລະນຳໃຊ້ໄດ້ຢ່າງຖືກຕ້ອງ. ໃນເລື່ອງດັ່ງກ່າວ, ມັນເປັນສິ່ງສໍາຄັນທີ່ຈະຮັກສາຄວາມເພິ່ງພໍໃຈຂອງຂໍ້ມູນ, ໂດຍສະເພາະຖ້າພວກເຂົາມາຈາກແຫຼ່ງຕ່າງໆ. ເຫຼົ່ານີ້ແມ່ນ, ແນ່ນອນ, ມາດຕະການບໍ່ພຽງພໍເພື່ອຮັກສາຄວາມສົມບູນຂອງຂໍ້ມູນ, ແຕ່ໃນບາງກໍລະນີມັນເປັນສິ່ງຈໍາເປັນ. ແມ່ນແລ້ວ, ແລະ idling ເພື່ອຄອບຄອງຊັບພະຍາກອນແມ່ນ irrational.

ດັ່ງນັ້ນ, ເຊັນເຊີຂອງພວກເຮົາຈະເປີດຕົວຕາມແນວຕັ້ງຕໍ່ມາຂອງກາຟຖ້າມີຂໍ້ມູນສົດຢູ່ໃນເມລ, ແລະຍັງຫມາຍຂໍ້ມູນທີ່ຜ່ານມາວ່າບໍ່ກ່ຽວຂ້ອງ.

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