कोई फर्क नहीं पड़ता कि तकनीक कितनी विकसित हो जाती है, पुराने दृष्टिकोणों की एक श्रृंखला हमेशा विकास के पीछे चलती है। यह एक सुचारु परिवर्तन, मानवीय कारकों, तकनीकी आवश्यकताओं, या कुछ और के कारण हो सकता है। डेटा प्रोसेसिंग के क्षेत्र में, इस भाग में डेटा स्रोत सबसे अधिक खुलासा करते हैं। इससे कोई फर्क नहीं पड़ता कि हम इससे छुटकारा पाने का कितना सपना देखते हैं, लेकिन अभी तक डेटा का हिस्सा तत्काल दूतों और ईमेलों में भेजा जाता है, अधिक पुरातन स्वरूपों का उल्लेख नहीं करना। मैं आपको अपाचे एयरफ्लो के विकल्पों में से एक को अलग करने के लिए आमंत्रित करता हूं, यह दर्शाता है कि आप ईमेल से डेटा कैसे ले सकते हैं।
पृष्ठभूमि
पारस्परिक संचार से लेकर कंपनियों के बीच बातचीत के मानकों तक, बहुत सारा डेटा अभी भी ई-मेल के माध्यम से स्थानांतरित किया जाता है। यह अच्छा है यदि डेटा प्राप्त करने के लिए एक इंटरफ़ेस लिखना संभव है या ऐसे लोगों को कार्यालय में रखना जो इस जानकारी को अधिक सुविधाजनक स्रोतों में दर्ज करेंगे, लेकिन अक्सर यह संभव नहीं हो पाता है। मुझे जिस विशिष्ट कार्य का सामना करना पड़ा, वह कुख्यात CRM सिस्टम को डेटा वेयरहाउस और फिर OLAP सिस्टम से जोड़ना था। ऐसा ऐतिहासिक रूप से हुआ है कि हमारी कंपनी के लिए व्यापार के किसी विशेष क्षेत्र में इस प्रणाली का उपयोग सुविधाजनक था। इसलिए, हर कोई वास्तव में इस तृतीय-पक्ष प्रणाली से भी डेटा के साथ काम करने में सक्षम होना चाहता था। सबसे पहले, बेशक, एक खुले एपीआई से डेटा प्राप्त करने की संभावना का अध्ययन किया गया। दुर्भाग्य से, एपीआई में सभी आवश्यक डेटा प्राप्त करना शामिल नहीं था, और, सरल शब्दों में, यह कई मायनों में टेढ़ा था, और तकनीकी सहायता अधिक व्यापक कार्यक्षमता प्रदान करने के लिए आधे रास्ते को पूरा नहीं करना चाहती थी या नहीं कर सकती थी। लेकिन इस प्रणाली ने संग्रह को अनलोड करने के लिए लिंक के रूप में मेल द्वारा समय-समय पर लापता डेटा प्राप्त करने का अवसर प्रदान किया।
यह ध्यान दिया जाना चाहिए कि यह एकमात्र ऐसा मामला नहीं था जिसमें व्यवसाय ईमेल या तत्काल दूतों से डेटा एकत्र करना चाहता था। हालाँकि, इस मामले में, हम किसी तृतीय-पक्ष कंपनी को प्रभावित नहीं कर सके जो केवल इस तरह से डेटा का हिस्सा प्रदान करती है।
अपाचे एयरफ्लो
ETL प्रक्रियाओं को बनाने के लिए, हम अक्सर Apache Airflow का उपयोग करते हैं। एक पाठक के लिए जो इस तकनीक से अपरिचित है, यह बेहतर ढंग से समझने के लिए कि यह संदर्भ में और सामान्य तौर पर कैसा दिखता है, मैं कुछ परिचयात्मक का वर्णन करूंगा।
Apache Airflow एक मुफ़्त प्लेटफ़ॉर्म है जिसका उपयोग Python में ETL (एक्सट्रैक्ट-ट्रांसफ़ॉर्म-लोडिंग) प्रक्रियाओं के निर्माण, निष्पादन और निगरानी के लिए किया जाता है। एयरफ्लो में मुख्य अवधारणा एक निर्देशित चक्रीय ग्राफ है, जहां ग्राफ के शीर्ष विशिष्ट प्रक्रियाएं हैं, और ग्राफ के किनारे नियंत्रण या सूचना का प्रवाह हैं। एक प्रक्रिया केवल किसी भी पायथन फ़ंक्शन को कॉल कर सकती है, या इसमें कक्षा के संदर्भ में अनुक्रमिक रूप से कई कार्यों को कॉल करने से अधिक जटिल तर्क हो सकते हैं। सबसे लगातार संचालन के लिए, पहले से ही कई तैयार विकास हैं जिनका उपयोग प्रक्रियाओं के रूप में किया जा सकता है। ऐसे विकासों में शामिल हैं:
- ऑपरेटर - डेटा को एक स्थान से दूसरे स्थान पर स्थानांतरित करने के लिए, उदाहरण के लिए, डेटाबेस तालिका से डेटा वेयरहाउस में;
- सेंसर - एक निश्चित घटना के होने की प्रतीक्षा करने और ग्राफ के बाद के शीर्षों पर नियंत्रण के प्रवाह को निर्देशित करने के लिए;
- हुक - निचले स्तर के संचालन के लिए, उदाहरण के लिए, डेटाबेस तालिका से डेटा प्राप्त करने के लिए (कथनों में प्रयुक्त);
- आदि
इस लेख में 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 के बारे में बताता हूँ। अपाचे एयरफ्लो कनेक्शन पैरामीटर (लॉगिन, पासवर्ड, पता और अन्य पैरामीटर) संग्रहीत करता है जिसे एक स्ट्रिंग पहचानकर्ता द्वारा एक्सेस किया जा सकता है। नेत्रहीन, कनेक्शन प्रबंधन इस तरह दिखता है
डेटा की प्रतीक्षा करने के लिए सेंसर
चूंकि हम पहले से ही जानते हैं कि मेल से डेटा कैसे कनेक्ट और प्राप्त करना है, अब हम उनके लिए प्रतीक्षा करने के लिए एक सेंसर लिख सकते हैं। मेरे मामले में, यह एक ऑपरेटर को तुरंत लिखने के लिए काम नहीं करता था जो डेटा को संसाधित करेगा, यदि कोई हो, क्योंकि अन्य प्रक्रियाएं मेल से प्राप्त डेटा के आधार पर काम करती हैं, जिसमें वे भी शामिल हैं जो अन्य स्रोतों (एपीआई, टेलीफोनी) से संबंधित डेटा लेते हैं। , वेब मेट्रिक्स, आदि) आदि)। मैं आपको एक उदाहरण देता हूँ। 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 में वापस, उन्होंने इसे पेश करने का वादा किया था, लेकिन जाहिर तौर पर उनका मन बदल गया। मैंने आवश्यक अक्षरों के लिए एक अलग फ़ोल्डर बनाकर और मेल वेब इंटरफ़ेस में आवश्यक अक्षरों के लिए फ़िल्टर स्थापित करके इस समस्या का समाधान किया। इस प्रकार, खोज के लिए केवल आवश्यक पत्र और शर्तें, मेरे मामले में, बस (अनदेखी) इस फ़ोल्डर में आती हैं।
संक्षेप में, हमारे पास निम्नलिखित अनुक्रम है: हम जाँचते हैं कि क्या नए अक्षर हैं जो शर्तों को पूरा करते हैं, यदि हैं, तो हम अंतिम पत्र से लिंक का उपयोग करके संग्रह को डाउनलोड करते हैं।
अंतिम बिंदुओं के तहत, यह छोड़ दिया गया है कि यह संग्रह अनपैक किया जाएगा, संग्रह से डेटा को साफ़ और संसाधित किया जाएगा, और परिणामस्वरूप, पूरी बात ईटीएल प्रक्रिया की पाइपलाइन में आगे बढ़ जाएगी, लेकिन यह पहले से ही परे है लेख का दायरा। यदि यह दिलचस्प और उपयोगी निकला, तो मैं खुशी से अपाचे एयरफ्लो के लिए ईटीएल समाधान और उनके भागों का वर्णन करना जारी रखूंगा।
स्रोत: www.habr.com