एयरफ्लो में डायनेमिक डीएजी जनरेशन

सभी को नमस्कार! मेरा नाम एंटोन है, रोस्टेलकॉम में मैं एक केंद्रीय डेटा वेयरहाउस विकसित कर रहा हूं। हमारे भंडारण में मॉड्यूल शामिल हैं, जिनमें से ऑर्केस्ट्रेटर कई Informatica उदाहरणों का उपयोग करता है, जिनमें से कुछ हम ओपन-सोर्स समाधानों के लिए संक्रमण के भाग के रूप में Airflow में स्थानांतरित करना चाहते हैं। चूंकि Informatica और Airflow मूल रूप से विभिन्न उपकरण हैं, इसलिए मौजूदा कार्यान्वयन को लेना और दोहराना इतना सरल नहीं है। हम एक वर्कफ़्लो प्राप्त करना चाहते थे, एक तरफ, वर्तमान कार्यान्वयन के लिए जितना संभव हो उतना करीब और दूसरी तरफ, सबसे दिलचस्प पहले एयरफ्लो सिद्धांत - डायनामिज्म का उपयोग करके, जो लचीलापन देता है।

इस छोटे लेख में, मैं Airflow में DAGs की वास्तव में गतिशील पीढ़ी के बारे में बात करना चाहता हूँ। इस विषय पर, इंटरनेट में मुख्य रूप से भारत के डेवलपर्स के कई लेख शामिल हैं, जो फ़ॉर्म की सामग्री हैं "आप एयरफ्लो में गतिशील रूप से डेग्स उत्पन्न कर सकते हैं, यहां एक उदाहरण है: <10 हैलोवर्ल्ड कार्यों / डैग्स> को उत्पन्न करने के लिए उदाहरण" । लेकिन हम डैग्स की पीढ़ी में रुचि रखते थे, जो कि एक चर संख्या और कार्यों के नामों के साथ समय में बदल जाएगी।

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

Habré पर Airflow पर Mail.ru के डेवलपर्स के कुछ लेख हैं जिनमें बुनियादी बातों का अच्छी तरह से वर्णन किया गया है:

एयरफ्लो का सामान्य विवरण
शाखाकरण, जिंजा के माध्यम से मानकीकरण, और एक्सकॉम के माध्यम से डीएजी के भीतर संचार

छोटी शब्दावली:

डीएजी / डीएजी एक निर्देशित एसाइक्लिक ग्राफ है। इस मामले में, हमारा मतलब उन क्रियाओं से है जो एक दूसरे पर निर्भर हैं और चक्र नहीं बनाते हैं।
SubDAG / सबडैग - DAG के समान, लेकिन एक अन्य DAG के अंदर स्थित, मूल DAG के भाग के रूप में लॉन्च किया गया (यानी, कार्य होने के नाते) और एक अलग शेड्यूल नहीं होने के कारण।
संचालक / संचालक - डग में एक विशिष्ट कदम, एक विशिष्ट क्रिया करते हुए। उदाहरण के लिए, पायथनऑपरेटर।
टास्क / टास्क - डीएजी शुरू करते समय ऑपरेटर का एक विशिष्ट उदाहरण, वेब इंटरफेस में एक छोटे वर्ग के रूप में कल्पना की जाती है। उदाहरण के लिए, PythonOperator, जिसे run_task कहा जाता है और DAG check_dag में चलता है।

डैग, समस्याओं और नुकसान में गतिशील कार्य पीढ़ी का विचार

इनपुट डेटा:

ऑर्केस्ट्रा रिपॉजिटरी में एक टेबल है, इसे PKG_TABLE कहते हैं।
एक तंत्र है जो PKG_TABLE तालिका में प्रविष्टियाँ जोड़ता है जो डेटा पैकेट डाउनलोड के लिए तैयार है।

हम क्या चाहते थे:

डीएजी, जो तैयार-से-डाउनलोड पैकेज के लिए उत्पन्न होगा और उन्हें डाउनलोड करना शुरू कर देगा (बिगाड़ने वाला: अंत में, सब कुछ निकला)।

नीचे दिए गए कोड का उपयोग करते हुए, हम LatestOnlyOperator कार्य और उसके आश्रित कार्य से मिलकर एक डेग उत्पन्न करते हैं, जो तब बनाया जाता है जब pkg_subdag_factory फ़ंक्शन चलाया जाता है, जो PKG_TABLE तालिका से संकुल की सूची प्राप्त करता है और कई PythonOperators उत्पन्न करता है। यदि डाउनलोड करने के लिए कोई पैकेज नहीं हैं, तो एक डमीऑपरेटर उत्पन्न होता है।

उन्होंने एक पायथनऑपरेटर के साथ पहला संस्करण बनाने का फैसला किया, इसे एयरफ्लो का उपयोग करके एक विस्तृत वर्कफ़्लो में पुनर्वितरित किया।

# -*- coding: utf-8 -*- """  DAG    """ from airflow.models import DAG from airflow.operators.python_operator import PythonOperator from airflow.operators.subdag_operator import SubDagOperator from airflow.operators.dummy_operator import DummyOperator from airflow.operators.latest_only_operator import LatestOnlyOperator from airflow.hooks.oracle_hook import OracleHook from datetime import datetime, timedelta import logging from scripts.lib import run_load, select_pkg_data def pkg_subdag_factory( oracle_hook, parent_dag_name, child_dag_name, start_date, schedule_interval, param_dict): """ ,  DAG    PythonOperator\` (1  - 1 PythonOperator)  : oracle_hook - airflow.hooks.oracle_hook.OracleHook parent_dag_name -  ""  child_dag_name -    start_date -       schedule_interval -      param_dict -     """ dag = DAG( '%s.%s' % (parent_dag_name, child_dag_name), schedule_interval=schedule_interval, start_date=start_date, catchup=False ) logging.info('selecting pkg data...') pkg_set = select_pkg_data(oracle_hook) if len(pkg_set): logging.info('pkg_set:') logging.info(pkg_set) for pkg in pkg_set: pkg_id = pkg[1] pkg_dict = {'pkg_data_' + str(pkg_id): pkg} param_dict.update(pkg_dict) task_name = 'pkg_' + str(pkg_id) PythonOperator( task_id=task_name, python_callable=run_load, op_kwargs={ 'oracle_hook': oracle_hook, 'param_dict': param_dict, 'pkg_id': pkg_id }, retries=0, dag=dag ) else: logging.info('Undelivered packages not found') DummyOperator(task_id='no_packages_dummy', retries=0, dag=dag) return dag interval = '*/10 * * * *' args = { 'owner': 'airflow', 'start_date': datetime(2018, 11, 12) } oracle_hook = OracleHook('ora_meta') main_dag_name = 'load' load_dag_name = 'load_packages' param_dict = { #       } main_dag = DAG( dag_id=main_dag_name, default_args=args, schedule_interval=interval, catchup=False ) subdag = SubDagOperator( subdag=pkg_subdag_factory( oracle_hook, main_dag_name, load_dag_name, args['start_date'], interval, param_dict ), task_id=load_dag_name, dag=main_dag ) #  ,       latest_only = LatestOnlyOperator(task_id='latest_only', dag=main_dag) subdag.set_upstream(latest_only) 

निम्नलिखित स्क्रीनशॉट दिखाते हैं कि यह परिणाम के रूप में कैसा दिखता है।
DAG की उपस्थिति:

वितरण के लिए संकुल की अनुपस्थिति में उपखंड की उपस्थिति:

वितरण के लिए संकुल की उपस्थिति में उपखंड का प्रकटन:

समस्याएं और वित्त

• कैचअप ने हमारी उम्मीद के मुताबिक काम नहीं किया: स्विच ऑफ डैग को चालू करने के बाद, कई लॉन्च हुए (शेड्यूल की पूरी अवधि के लिए नहीं, लेकिन एक ही समय में 2-3)। इस वजह से, मुझे LatestOnlyOperator को जोड़ना पड़ा, ताकि पिछले एक को छोड़कर सभी लॉन्च बेकार हो जाएं।
• यदि आप एक सबडैग बनाते हैं, तो आपको कमांड लाइन के माध्यम से इसे स्पष्ट रूप से सक्षम करने की आवश्यकता है कमांड "airflow unpause <subdag_name>", अन्यथा यह शुरू नहीं होता है, और प्रत्येक नया सबडैग (नए नाम के साथ एक सबडैग) बनाते समय आपको ऐसा करने की आवश्यकता होती है, जो गतिशील रूप से उत्पन्न करने के लिए बहुत असुविधाजनक होगा। । यदि आप "dags_are_paused_at_creation" पैरामीटर को एयरफ़्लो कॉन्फ़िगरेशन ($ airflow_home / airflow.cfg) में सेट करते हैं, तो यह आवश्यक नहीं होगा, लेकिन यह एक नए डेग के स्वत: स्वत: लॉन्च के साथ अप्रिय परिणाम दे सकता है - यह मुझे लगता है कि आपको नए डैग्स को मैन्युअल रूप से शुरू करने की आवश्यकता है।

जैसा कि प्रलेखन में कहा गया है, "एयरफ्लो की एक प्रमुख क्षमता यह है कि ये डीएजी रन परमाणु, बेकार वस्तुएं, <...>" हैं, जिसका अर्थ है: "यह समझा जाता है कि डैग अपरिवर्तित है।" इस तथ्य के कारण कि हमने इस "प्रमुख क्षमता" का उल्लंघन किया, हमने कुछ चीजें सीखीं:

• एक खाली डैग (कार्यों के बिना) शुरू होता है और समाप्त नहीं हो सकता है, सभी संभव समानताएं रोकते हुए। यह उस समय हुआ जब डाॅग शुरू होने के समय कोई डाउनलोड पैकेज नहीं थे। इसके चारों ओर काम करने के लिए, एक डमीपरेटर बनाया जाता है।
• अगर काम के दौरान टास्क डेग रीजेनर हो जाता है और अपडेटेड डैग में यह काम नहीं रह जाता है - तो यह रनिंग प्रोसेस में रुकावट के साथ बंद हो जाएगा। और यह शेड्यूलर के प्रत्येक चरण के साथ होता है, लेकिन एयरफ्लो कॉन्फ़िगरेशन ($ airflow_home / airflow.cfg) में min_file_process_interval पैरामीटर से अधिक बार इंगित नहीं किया जाता है। इसे दरकिनार करने के लिए, हमने न केवल "डाउनलोड करने के लिए तैयार" की स्थिति से, बल्कि "लोडिंग इन प्रोसेस" की स्थिति से भी टास्क पैक्स का निर्माण किया, ताकि डाउनलोड जारी रहने के दौरान यह उत्पन्न होता रहे।
• यदि डैग के वर्तमान संस्करण में कोई कार्य नहीं है जो पहले था - उदाहरण के लिए, नाम "pkg_123" के साथ एक कार्य था जिसे पहले लोड किया गया था और डेग के वर्तमान संस्करण में नहीं बनाया गया है, तो आप वेब इंटरफ़ेस में इस कार्य के आँकड़े नहीं देख सकते हैं। यद्यपि सभी जानकारी एयरफ्लो डेटाबेस में संग्रहीत है और इसके आधार पर बाहरी साधनों द्वारा पुराने लॉन्च के लिए एक सुंदर डैशबोर्ड बनाना संभव है। जब सवाल डीएजी को अपडेट करने की आवृत्ति और इसे अक्षम करने की क्षमता के बारे में उठता है, तो आप इसके बारे में यहां पढ़ सकते हैं।
• टास्क_ड की गतिशील पीढ़ी के कारण, प्रत्येक मौजूदा कार्य के लिए डेटा के साथ एक शब्दकोश को फेंकना आवश्यक है, साथ ही ऐसे प्रत्येक कार्य में, वर्तमान पैकेज की आईडी, ताकि जब फ़ंक्शन स्वयं काम करता है, पैकेज आईडी द्वारा उसी शब्दकोश से आवश्यक डेटा का चयन करें। अन्यथा, एक ही पैकेज के लिए सभी कार्य शुरू हो गए।

Execution_date लॉग और वास्तविक प्रारंभ समय में

मैं एयरफ्लो की एक और बारीकियों के साथ समाप्त हो जाऊंगा, जो पहले भ्रमित करता है और अन्य लेखों में सरल शब्दों में वर्णित नहीं किया जाता है - निष्पादन_डेट (जो सभी लॉग में प्रदर्शित होता है, इंटरफ़ेस आदि) और वास्तविक प्रारंभ समय। सिद्धांत रूप में, वर्णन एयरफ्लो और एफएक्यू के प्रलेखन में है , लेकिन परिणाम स्पष्ट नहीं है, इसलिए यह मुझे लगता है कि स्पष्टीकरण की आवश्यकता है।

दस्तावेज़ीकरण : "शेड्यूलर ने अवधि के अंत में अपनी नौकरी शुरू की"
परिणाम : यदि आप एक शेड्यूल के साथ एक डाग बनाते हैं, उदाहरण के लिए, @daily, तो निष्पादन के साथ एक रन "2018-01-01 00:00:00" वास्तव में "2018-02-01 00:00:00" चलेगा।

उपयोगी लिंक:

कैचअप प्रलेखन
LatestOnlyOperator प्रलेखन
LatestOnlyOperator पर अधिक प्रलेखन
LatestOnlyOperator का उदाहरण उपयोग
कुछ बारीकियाँ
पिछले लॉन्च पर निर्भरता के बारे में प्रश्न
गतिशील पीढ़ी के बारे में एक छोटा सा उदाहरण
छोटे विवरण के साथ गतिशील पीढ़ी के बारे में एक प्रश्न