🥇एयरफ़्लो बैच डेटा प्रोसेसिंग प्रक्रियाओं को आसानी से और तेज़ी से विकसित करने और समर्थन करने का एक उपकरण है

नमस्ते, हबर! इस लेख में मैं बैच डेटा प्रोसेसिंग प्रक्रियाओं को विकसित करने के लिए एक महान उपकरण के बारे में बात करना चाहता हूं, उदाहरण के लिए, कॉर्पोरेट डीडब्ल्यूएच या आपके डेटालेक के बुनियादी ढांचे में। हम अपाचे एयरफ्लो (इसके बाद एयरफ्लो के रूप में संदर्भित) के बारे में बात करेंगे। यह गलत तरीके से हैबे पर ध्यान देने से वंचित है, और मुख्य भाग में मैं आपको यह समझाने की कोशिश करूंगा कि आपकी ईटीएल/ईएलटी प्रक्रियाओं के लिए शेड्यूलर चुनते समय कम से कम एयरफ्लो देखने लायक है।

इससे पहले, जब मैं टिंकॉफ बैंक में काम करता था तो मैंने डीडब्ल्यूएच के विषय पर लेखों की एक श्रृंखला लिखी थी। अब मैं Mail.Ru Group टीम का हिस्सा बन गया हूं और गेमिंग क्षेत्र में डेटा विश्लेषण के लिए एक प्लेटफॉर्म विकसित कर रहा हूं। दरअसल, जैसे ही समाचार और दिलचस्प समाधान सामने आएंगे, मैं और मेरी टीम यहां डेटा एनालिटिक्स के लिए हमारे प्लेटफॉर्म के बारे में बात करेंगे।

प्रस्तावना

तो, चलिए शुरू करते हैं। वायुप्रवाह क्या है? यह एक पुस्तकालय है (या पुस्तकालयों का सेट) कार्य प्रक्रियाओं का विकास, योजना और निगरानी करना। एयरफ़्लो की मुख्य विशेषता: पायथन कोड का उपयोग प्रक्रियाओं का वर्णन (विकसित) करने के लिए किया जाता है। आपके प्रोजेक्ट और विकास को व्यवस्थित करने के लिए इसके बहुत सारे फायदे हैं: संक्षेप में, आपका (उदाहरण के लिए) ईटीएल प्रोजेक्ट सिर्फ एक पायथन प्रोजेक्ट है, और आप बुनियादी ढांचे, टीम के आकार और विशिष्टताओं को ध्यान में रखते हुए इसे अपनी इच्छानुसार व्यवस्थित कर सकते हैं। अन्य आवश्यकताएं। वाद्ययंत्र की दृष्टि से सब कुछ सरल है। उदाहरण के लिए PyCharm + Git का उपयोग करें। यह अद्भुत और बहुत सुविधाजनक है!

आइए अब एयरफ़्लो की मुख्य संस्थाओं पर नज़र डालें। उनके सार और उद्देश्य को समझकर, आप अपनी प्रक्रिया वास्तुकला को बेहतर ढंग से व्यवस्थित कर सकते हैं। शायद मुख्य इकाई डायरेक्टेड एसाइक्लिक ग्राफ़ (इसके बाद डीएजी के रूप में संदर्भित) है।

डेग

डीएजी आपके कार्यों का कुछ सार्थक जुड़ाव है जिसे आप एक विशिष्ट कार्यक्रम के अनुसार कड़ाई से परिभाषित अनुक्रम में पूरा करना चाहते हैं। एयरफ़्लो डीएजी और अन्य संस्थाओं के साथ काम करने के लिए एक सुविधाजनक वेब इंटरफ़ेस प्रदान करता है:

डीएजी इस तरह दिख सकता है:

डेवलपर, DAG डिज़ाइन करते समय, ऑपरेटरों का एक सेट निर्धारित करता है, जिस पर DAG के भीतर कार्य बनाए जाएंगे। यहां हम एक अन्य महत्वपूर्ण इकाई पर आते हैं: एयरफ्लो ऑपरेटर।

ऑपरेटरों

ऑपरेटर एक इकाई है जिसके आधार पर जॉब इंस्टेंस बनाए जाते हैं, जो बताता है कि जॉब इंस्टेंस के निष्पादन के दौरान क्या होगा। GitHub से एयरफ़्लो रिलीज़ होता है इसमें पहले से ही उपयोग के लिए तैयार ऑपरेटरों का एक सेट मौजूद है। उदाहरण:

बैशऑपरेटर - बैश कमांड निष्पादित करने के लिए ऑपरेटर।
PythonOperator - Python कोड को कॉल करने के लिए ऑपरेटर।
ईमेलऑपरेटर - ईमेल भेजने के लिए ऑपरेटर।
HTTPऑपरेटर - http अनुरोधों के साथ काम करने के लिए ऑपरेटर।
SqlOperator - SQL कोड निष्पादित करने के लिए ऑपरेटर।
सेंसर किसी घटना की प्रतीक्षा करने के लिए एक ऑपरेटर है (आवश्यक समय का आगमन, आवश्यक फ़ाइल की उपस्थिति, डेटाबेस में एक लाइन, एपीआई से प्रतिक्रिया, आदि, आदि)।

अधिक विशिष्ट ऑपरेटर हैं: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator।

आप अपनी विशेषताओं के आधार पर ऑपरेटर भी विकसित कर सकते हैं और उन्हें अपने प्रोजेक्ट में उपयोग कर सकते हैं। उदाहरण के लिए, हमने MongoDBToHiveViaHdfsTransfer बनाया, जो MongoDB से Hive में दस्तावेज़ निर्यात करने के लिए एक ऑपरेटर है, और इसके साथ काम करने के लिए कई ऑपरेटर हैं। क्लिकहाउस: CHLoadFromHiveOperator और CHTableLoaderOperator। अनिवार्य रूप से, जैसे ही किसी प्रोजेक्ट में मूल कथनों पर निर्मित कोड का बार-बार उपयोग किया जाता है, आप इसे एक नए कथन में बनाने के बारे में सोच सकते हैं। इससे आगे का विकास सरल हो जाएगा, और आप प्रोजेक्ट में ऑपरेटरों की अपनी लाइब्रेरी का विस्तार करेंगे।

इसके बाद, कार्यों के इन सभी उदाहरणों को निष्पादित करने की आवश्यकता है, और अब हम शेड्यूलर के बारे में बात करेंगे।

समयबद्धक

एयरफ्लो का कार्य शेड्यूलर पर बनाया गया है अजवाइन. सेलेरी एक पायथन लाइब्रेरी है जो आपको कतार के साथ-साथ अतुल्यकालिक और वितरित कार्यों को व्यवस्थित करने की अनुमति देती है। एयरफ़्लो पक्ष पर, सभी कार्यों को पूल में विभाजित किया गया है। पूल मैन्युअल रूप से बनाए जाते हैं। आमतौर पर, उनका उद्देश्य स्रोत के साथ काम करने के कार्यभार को सीमित करना या डीडब्ल्यूएच के भीतर कार्यों को टाइप करना है। पूल को वेब इंटरफ़ेस के माध्यम से प्रबंधित किया जा सकता है:

प्रत्येक पूल में स्लॉट की संख्या की एक सीमा होती है। DAG बनाते समय, इसे एक पूल दिया जाता है:

ALERT_MAILS =  Variable.get("gv_mail_admin_dwh")
DAG_NAME = 'dma_load'
OWNER = 'Vasya Pupkin'
DEPENDS_ON_PAST = True
EMAIL_ON_FAILURE = True
EMAIL_ON_RETRY = True
RETRIES = int(Variable.get('gv_dag_retries'))
POOL = 'dma_pool'
PRIORITY_WEIGHT = 10

start_dt = datetime.today() - timedelta(1)
start_dt = datetime(start_dt.year, start_dt.month, start_dt.day)

default_args = {
    'owner': OWNER,
    'depends_on_past': DEPENDS_ON_PAST,
    'start_date': start_dt,
    'email': ALERT_MAILS,
    'email_on_failure': EMAIL_ON_FAILURE,
    'email_on_retry': EMAIL_ON_RETRY,
    'retries': RETRIES,
    'pool': POOL,
    'priority_weight': PRIORITY_WEIGHT
}
dag = DAG(DAG_NAME, default_args=default_args)
dag.doc_md = __doc__

DAG स्तर पर परिभाषित पूल को कार्य स्तर पर ओवरराइड किया जा सकता है।
एक अलग प्रक्रिया, शेड्यूलर, एयरफ्लो में सभी कार्यों को शेड्यूल करने के लिए जिम्मेदार है। दरअसल, शेड्यूलर निष्पादन के लिए कार्यों को निर्धारित करने के सभी तंत्रों से संबंधित है। कार्य निष्पादित होने से पहले कई चरणों से गुजरता है:

डीएजी में पिछले कार्य पूरे हो चुके हैं; एक नया कार्य कतारबद्ध किया जा सकता है।
कतार को कार्यों की प्राथमिकता के आधार पर क्रमबद्ध किया जाता है (प्राथमिकताओं को भी नियंत्रित किया जा सकता है), और यदि पूल में एक मुफ्त स्लॉट है, तो कार्य को संचालन में लिया जा सकता है।
यदि कोई निःशुल्क श्रमिक अजवाइन है, तो कार्य उसे भेजा जाता है; समस्या में आपने जो कार्य प्रोग्राम किया है वह एक या दूसरे ऑपरेटर का उपयोग करके शुरू होता है।

काफी सरल है।

शेड्यूलर सभी डीएजी के सेट और डीएजी के भीतर सभी कार्यों पर चलता है।

शेड्यूलर को डीएजी के साथ काम शुरू करने के लिए, डीएजी को एक शेड्यूल निर्धारित करना होगा:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='@hourly')

तैयार प्रीसेट का एक सेट है: @once, @hourly, @daily, @weekly, @monthly, @yearly.

आप क्रॉन अभिव्यक्तियों का भी उपयोग कर सकते हैं:

dag = DAG(DAG_NAME, default_args=default_args, schedule_interval='*/10 * * * *')

कार्यवाही की तारीख

यह समझने के लिए कि एयरफ्लो कैसे काम करता है, यह समझना महत्वपूर्ण है कि डीएजी के लिए निष्पादन तिथि क्या है। एयरफ़्लो में, DAG का एक निष्पादन दिनांक आयाम होता है, अर्थात, DAG के कार्य शेड्यूल के आधार पर, प्रत्येक निष्पादन तिथि के लिए कार्य उदाहरण बनाए जाते हैं। और प्रत्येक निष्पादन तिथि के लिए, कार्यों को फिर से निष्पादित किया जा सकता है - या, उदाहरण के लिए, एक डीएजी कई निष्पादन तिथियों में एक साथ काम कर सकता है। यह यहाँ स्पष्ट रूप से दिखाया गया है:

दुर्भाग्य से (या शायद सौभाग्य से: यह स्थिति पर निर्भर करता है), यदि डीएजी में कार्य के कार्यान्वयन को सही किया जाता है, तो पिछली निष्पादन तिथि में निष्पादन समायोजन को ध्यान में रखते हुए आगे बढ़ेगा। यह अच्छा है यदि आपको एक नए एल्गोरिथ्म का उपयोग करके पिछली अवधि में डेटा की पुनर्गणना करने की आवश्यकता है, लेकिन यह बुरा है क्योंकि परिणाम की पुनरुत्पादकता खो गई है (बेशक, कोई भी आपको Git से स्रोत कोड के आवश्यक संस्करण को वापस करने और गणना करने के लिए परेशान नहीं करता है) आपको एक समय की आवश्यकता है, जिस तरह से आपको इसकी आवश्यकता है)।

कार्य उत्पन्न करना

डीएजी का कार्यान्वयन पायथन में कोड है, इसलिए हमारे पास काम करते समय कोड की मात्रा को कम करने का एक बहुत ही सुविधाजनक तरीका है, उदाहरण के लिए, शार्ड स्रोतों के साथ। मान लीजिए कि आपके पास स्रोत के रूप में तीन MySQL शार्ड हैं, आपको प्रत्येक पर चढ़ने और कुछ डेटा लेने की आवश्यकता है। इसके अलावा, स्वतंत्र रूप से और समानांतर में। DAG में पायथन कोड इस तरह दिख सकता है:

connection_list = lv.get('connection_list')

export_profiles_sql = '''
SELECT
  id,
  user_id,
  nickname,
  gender,
  {{params.shard_id}} as shard_id
FROM profiles
'''

for conn_id in connection_list:
    export_profiles = SqlToHiveViaHdfsTransfer(
        task_id='export_profiles_from_' + conn_id,
        sql=export_profiles_sql,
        hive_table='stg.profiles',
        overwrite=False,
        tmpdir='/data/tmp',
        conn_id=conn_id,
        params={'shard_id': conn_id[-1:], },
        compress=None,
        dag=dag
    )
    export_profiles.set_upstream(exec_truncate_stg)
    export_profiles.set_downstream(load_profiles)

डीएजी इस तरह दिखता है:

इस मामले में, आप केवल सेटिंग्स को समायोजित करके और डीएजी को अपडेट करके एक शार्ड जोड़ या हटा सकते हैं। आरामदायक!

आप अधिक जटिल कोड पीढ़ी का भी उपयोग कर सकते हैं, उदाहरण के लिए, डेटाबेस के रूप में स्रोतों के साथ काम करें या तालिका संरचना का वर्णन करें, तालिका के साथ काम करने के लिए एक एल्गोरिदम, और, डीडब्ल्यूएच बुनियादी ढांचे की विशेषताओं को ध्यान में रखते हुए, एक प्रक्रिया उत्पन्न करें आपके स्टोरेज में एन टेबल लोड करने के लिए। या, उदाहरण के लिए, एक एपीआई के साथ काम करना जो सूची के रूप में पैरामीटर के साथ काम करने का समर्थन नहीं करता है, आप इस सूची से डीएजी में एन कार्य उत्पन्न कर सकते हैं, एपीआई में अनुरोधों की समानता को एक पूल तक सीमित कर सकते हैं और स्क्रैप कर सकते हैं एपीआई से आवश्यक डेटा. लचीला!

कोष

एयरफ़्लो का अपना बैकएंड रिपॉजिटरी है, एक डेटाबेस (MySQL या Postgres हो सकता है, हमारे पास Postgres है), जो कार्यों की स्थिति, DAGs, कनेक्शन सेटिंग्स, वैश्विक चर इत्यादि को संग्रहीत करता है। यहां मैं चाहूंगा कि मैं यह कह सकूं कि यदि आप इसके ऊपर अपनी कोई प्रक्रिया बनाना चाहते हैं तो एयरफ्लो में रिपॉजिटरी बहुत सरल (लगभग 20 टेबल) और सुविधाजनक है। मुझे इन्फॉर्मेटिका रिपॉजिटरी में 100500 तालिकाएँ याद हैं, जिन्हें क्वेरी बनाने का तरीका समझने से पहले लंबे समय तक अध्ययन करना पड़ता था।

निगरानी

रिपॉजिटरी की सरलता को देखते हुए, आप एक कार्य निगरानी प्रक्रिया बना सकते हैं जो आपके लिए सुविधाजनक हो। हम ज़ेपेलिन में एक नोटपैड का उपयोग करते हैं, जहां हम कार्यों की स्थिति देखते हैं:

यह एयरफ़्लो का वेब इंटरफ़ेस भी हो सकता है:

एयरफ़्लो कोड खुला स्रोत है, इसलिए हमने टेलीग्राम में अलर्ट जोड़ा है। किसी कार्य का प्रत्येक चल रहा उदाहरण, यदि कोई त्रुटि होती है, तो टेलीग्राम में समूह को स्पैम कर देता है, जहां संपूर्ण विकास और सहायता टीम शामिल होती है।

हमें टेलीग्राम के माध्यम से त्वरित प्रतिक्रिया मिलती है (यदि आवश्यक हो), और ज़ेपेलिन के माध्यम से हमें एयरफ़्लो में कार्यों की एक समग्र तस्वीर प्राप्त होती है।

कुल मिलाकर

एयरफ़्लो मुख्य रूप से खुला स्रोत है, और आपको इससे चमत्कार की उम्मीद नहीं करनी चाहिए। कारगर समाधान बनाने में समय और प्रयास लगाने के लिए तैयार रहें। लक्ष्य प्राप्त करने योग्य है, मेरा विश्वास करो, यह इसके लायक है। विकास की गति, लचीलापन, नई प्रक्रियाओं को जोड़ने में आसानी - आपको यह पसंद आएगा। बेशक, आपको परियोजना के संगठन, एयरफ्लो की स्थिरता पर बहुत अधिक ध्यान देने की आवश्यकता है: चमत्कार नहीं होते हैं।

अब हमारे पास एयरफ्लो प्रतिदिन काम कर रहा है लगभग 6,5 हजार कार्य. वे चरित्र में काफी भिन्न हैं। कई अलग-अलग और बहुत विशिष्ट स्रोतों से मुख्य DWH में डेटा लोड करने के कार्य हैं, मुख्य DWH के अंदर स्टोरफ्रंट की गणना करने के कार्य हैं, तेज़ DWH में डेटा प्रकाशित करने के कार्य हैं, कई, कई अलग-अलग कार्य हैं - और एयरफ्लो दिन-प्रतिदिन उन सबको चबाता है। संख्या में कहें तो यह है 2,3 हजारों DWH (Hadoop) के भीतर अलग-अलग जटिलता के ELT कार्य, लगभग। 2,5 सौ डेटाबेस सूत्रों के अनुसार, यह एक टीम है 4 ईटीएल डेवलपर्स, जो डीडब्ल्यूएच में ईटीएल डेटा प्रोसेसिंग और डीडब्ल्यूएच के अंदर ईएलटी डेटा प्रोसेसिंग और निश्चित रूप से अधिक में विभाजित हैं एक व्यवस्थापक, जो सेवा के बुनियादी ढांचे से संबंधित है।

Планы на будущее

प्रक्रियाओं की संख्या अनिवार्य रूप से बढ़ रही है, और एयरफ्लो बुनियादी ढांचे के संदर्भ में हम जो मुख्य काम करेंगे वह स्केलिंग है। हम एक एयरफ्लो क्लस्टर बनाना चाहते हैं, सेलेरी श्रमिकों के लिए पैरों की एक जोड़ी आवंटित करना चाहते हैं, और नौकरी शेड्यूलिंग प्रक्रियाओं और एक रिपॉजिटरी के साथ एक स्व-डुप्लिकेटिंग हेड बनाना चाहते हैं।

उपसंहार

बेशक, यह वह सब कुछ नहीं है जो मैं एयरफ्लो के बारे में बताना चाहता हूं, लेकिन मैंने मुख्य बिंदुओं पर प्रकाश डालने की कोशिश की है। खाने से भूख लगती है, इसे आज़माएं और आपको यह पसंद आएगा :)

स्रोत: www.habr.com