🥇एअरफ्लो हे बॅच डेटा प्रोसेसिंग प्रक्रियांना सोयीस्करपणे आणि त्वरीत विकसित आणि समर्थन देण्यासाठी एक साधन आहे

हॅलो, हॅब्र! या लेखात मला बॅच डेटा प्रोसेसिंग प्रक्रिया विकसित करण्यासाठी एका उत्तम साधनाबद्दल बोलायचे आहे, उदाहरणार्थ, कॉर्पोरेट डीडब्ल्यूएच किंवा तुमच्या डेटालेकच्या पायाभूत सुविधांमध्ये. आम्ही Apache Airflow (यापुढे Airflow म्हणून संदर्भित) बद्दल बोलू. हे Habré वर अन्यायकारकपणे लक्ष देण्यापासून वंचित आहे आणि मुख्य भागात मी तुम्हाला हे पटवून देण्याचा प्रयत्न करेन की तुमच्या ETL/ELT प्रक्रियेसाठी शेड्युलर निवडताना किमान Airflow पाहण्यासारखे आहे.

पूर्वी, मी टिंकॉफ बँकेत काम करताना DWH विषयावर लेखांची मालिका लिहिली होती. आता मी Mail.Ru ग्रुप टीमचा भाग झालो आहे आणि गेमिंग क्षेत्रात डेटा विश्लेषणासाठी एक व्यासपीठ विकसित करत आहे. वास्तविक, बातम्या आणि मनोरंजक उपाय दिसू लागल्यावर, मी आणि माझी टीम डेटा विश्लेषणासाठी आमच्या प्लॅटफॉर्मबद्दल येथे बोलू.

प्रस्तावना

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

आता एअरफ्लोच्या मुख्य घटकांकडे पाहू. त्यांचे सार आणि उद्देश समजून घेऊन, तुम्ही तुमची प्रक्रिया आर्किटेक्चर चांगल्या प्रकारे व्यवस्थित करू शकता. कदाचित मुख्य घटक डायरेक्टेड अॅसायक्लिक ग्राफ (यापुढे DAG म्हणून संदर्भित) आहे.

दिवस

डीएजी ही तुमच्या कार्यांची काही अर्थपूर्ण संघटना आहे जी तुम्ही विशिष्ट वेळापत्रकानुसार काटेकोरपणे परिभाषित क्रमाने पूर्ण करू इच्छिता. एअरफ्लो DAGs आणि इतर घटकांसह कार्य करण्यासाठी सोयीस्कर वेब इंटरफेस प्रदान करते:

DAG यासारखे दिसू शकते:

विकासक, डीएजीची रचना करताना, ऑपरेटरचा एक संच ठेवतो ज्यावर DAG अंतर्गत कार्ये तयार केली जातील. येथे आपण आणखी एका महत्त्वाच्या घटकाकडे आलो आहोत: एअरफ्लो ऑपरेटर.

ऑपरेटर

ऑपरेटर ही एक संस्था आहे ज्याच्या आधारावर नोकरीची उदाहरणे तयार केली जातात, जी नोकरीच्या उदाहरणाच्या अंमलबजावणीदरम्यान काय घडेल याचे वर्णन करते. GitHub वरून एअरफ्लो रिलीज होतो आधीपासून वापरण्यासाठी सज्ज ऑपरेटर्सचा संच आहे. उदाहरणे:

बॅशऑपरेटर - बॅश कमांड कार्यान्वित करण्यासाठी ऑपरेटर.
PythonOperator - Python कोड कॉल करण्यासाठी ऑपरेटर.
EmailOperator — ईमेल पाठवण्यासाठी ऑपरेटर.
HTTPOperator - HTTP विनंत्यांसह कार्य करण्यासाठी ऑपरेटर.
SqlOperator - SQL कोड कार्यान्वित करण्यासाठी ऑपरेटर.
सेन्सर हा इव्हेंटची वाट पाहणारा ऑपरेटर आहे (आवश्यक वेळेचे आगमन, आवश्यक फाइलचे स्वरूप, डेटाबेसमधील एक ओळ, API कडून प्रतिसाद इ. इ.).

अधिक विशिष्ट ऑपरेटर आहेत: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator.

तुम्ही तुमच्या स्वतःच्या वैशिष्ट्यांवर आधारित ऑपरेटर विकसित करू शकता आणि त्यांचा तुमच्या प्रकल्पात वापर करू शकता. उदाहरणार्थ, आम्ही MongoDBToHiveViaHdfsTransfer, मोंगोडीबी वरून Hive मध्ये दस्तऐवज निर्यात करण्यासाठी ऑपरेटर आणि काम करण्यासाठी अनेक ऑपरेटर तयार केले क्लिकहाऊस: CHLoadFromHiveOperator आणि CHTableLoaderOperator. मूलत:, एखाद्या प्रोजेक्टमध्ये बेसिक स्टेटमेंटवर बनवलेला कोड वारंवार वापरला गेल्यावर, तुम्ही ते नवीन स्टेटमेंटमध्ये बनवण्याचा विचार करू शकता. हे पुढील विकास सुलभ करेल आणि तुम्ही प्रकल्पातील ऑपरेटर्सची तुमची लायब्ररी विस्तारित कराल.

पुढे, कार्यांची ही सर्व उदाहरणे कार्यान्वित करणे आवश्यक आहे आणि आता आपण शेड्यूलरबद्दल बोलू.

शेड्युलर

एअरफ्लोचे टास्क शेड्युलर तयार केले आहे सफरचंद. सेलेरी ही एक पायथन लायब्ररी आहे जी तुम्हाला रांग आणि अतुल्यकालिक आणि वितरित कार्यांची अंमलबजावणी करण्यास अनुमती देते. एअरफ्लो बाजूला, सर्व कार्ये पूलमध्ये विभागली जातात. पूल स्वहस्ते तयार केले जातात. सामान्यतः, त्यांचा उद्देश स्त्रोतासह काम करण्याचा वर्कलोड मर्यादित करणे किंवा DWH मधील कार्ये टाइप करणे हा आहे. पूल वेब इंटरफेसद्वारे व्यवस्थापित केले जाऊ शकतात:

प्रत्येक पूलमध्ये स्लॉटच्या संख्येवर मर्यादा असते. 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 मध्ये पूर्वीची कामे पूर्ण झाली आहेत; नवीन कार्य रांगेत लावले जाऊ शकते.
कार्यांच्या प्राधान्यावर अवलंबून रांगेची क्रमवारी लावली जाते (प्राधान्यक्रम देखील नियंत्रित केले जाऊ शकतात), आणि जर पूलमध्ये एक विनामूल्य स्लॉट असेल, तर कार्य कार्यान्वित केले जाऊ शकते.
जर विनामूल्य कार्यकर्ता सेलेरी असेल तर, कार्य त्याच्याकडे पाठवले जाते; आपण समस्येमध्ये प्रोग्राम केलेले कार्य एक किंवा दुसर्या ऑपरेटरचा वापर करून सुरू होते.

पुरेशी साधी.

शेड्युलर सर्व DAGs आणि DAGs मधील सर्व कार्यांच्या सेटवर चालते.

शेड्युलरने DAG सोबत काम सुरू करण्‍यासाठी, 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 ला एक एक्झिक्यूशन डेट डायमेंशन असते, म्हणजे, DAG च्या कामाच्या शेड्यूलनुसार, प्रत्येक एक्झिक्यूशन डेटसाठी टास्क इन्स्टन्स तयार केले जातात. आणि प्रत्येक अंमलबजावणीच्या तारखेसाठी, कार्ये पुन्हा कार्यान्वित केली जाऊ शकतात - किंवा, उदाहरणार्थ, DAG अनेक अंमलबजावणी तारखांमध्ये एकाच वेळी कार्य करू शकते. हे येथे स्पष्टपणे दर्शविले आहे:

दुर्दैवाने (किंवा कदाचित सुदैवाने: ते परिस्थितीवर अवलंबून असते), जर DAG मधील कार्याची अंमलबजावणी दुरुस्त केली गेली, तर मागील अंमलबजावणीच्या तारखेची अंमलबजावणी समायोजने लक्षात घेऊन पुढे जाईल. जर तुम्हाला नवीन अल्गोरिदम वापरून मागील कालावधीतील डेटाची पुनर्गणना करायची असेल तर हे चांगले आहे, परंतु ते वाईट आहे कारण परिणामाची पुनरुत्पादन क्षमता गमावली आहे (अर्थातच, Git वरून स्त्रोत कोडची आवश्यक आवृत्ती परत करण्यास कोणीही तुम्हाला त्रास देत नाही आणि काय गणना करा. आपल्याला एक वेळ हवा आहे, ज्या प्रकारे आपल्याला त्याची आवश्यकता आहे).

कार्ये निर्माण करणे

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

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)

डीएजी असे दिसते:

या प्रकरणात, तुम्ही फक्त सेटिंग्ज समायोजित करून आणि DAG अद्यतनित करून शार्ड जोडू किंवा काढू शकता. आरामदायक!

आपण अधिक जटिल कोड जनरेशन देखील वापरू शकता, उदाहरणार्थ, डेटाबेसच्या स्वरूपात स्त्रोतांसह कार्य करा किंवा टेबल स्ट्रक्चरचे वर्णन करा, टेबलसह कार्य करण्यासाठी अल्गोरिदम आणि DWH इन्फ्रास्ट्रक्चरची वैशिष्ट्ये विचारात घेऊन प्रक्रिया तयार करा. तुमच्या स्टोरेजमध्ये N टेबल लोड करण्यासाठी. किंवा, उदाहरणार्थ, एपीआयसह कार्य करणे जे सूचीच्या स्वरूपात पॅरामीटरसह कार्य करण्यास समर्थन देत नाही, तुम्ही या सूचीमधून DAG मध्ये N कार्ये व्युत्पन्न करू शकता, API मधील विनंत्यांची समांतरता पूलमध्ये मर्यादित करू शकता आणि स्क्रॅप करू शकता. API कडून आवश्यक डेटा. लवचिक!

भांडार

एअरफ्लोचे स्वतःचे बॅकएंड रेपॉजिटरी आहे, एक डेटाबेस (मायएसक्यूएल किंवा पोस्टग्रेस असू शकतो, आमच्याकडे पोस्टग्रेस आहे), जे कार्यांची स्थिती, डीएजी, कनेक्शन सेटिंग्ज, ग्लोबल व्हेरिएबल्स इ. इत्यादी संग्रहित करते. येथे मला असे म्हणायचे आहे की एअरफ्लो मधील रेपॉजिटरी अगदी सोपी आहे (सुमारे 20 टेबल्स) आणि जर तुम्हाला तुमच्या स्वतःच्या कोणत्याही प्रक्रिया तयार करायच्या असतील तर ते सोयीस्कर आहे. मला इन्फॉर्मेटिका रेपॉजिटरीमधील 100500 टेबल्स आठवतात, ज्याचा प्रश्न कसा तयार करायचा हे समजून घेण्यापूर्वी बराच काळ अभ्यास करावा लागला.

देखरेख

रेपॉजिटरीची साधेपणा लक्षात घेता, तुम्ही तुमच्यासाठी सोयीस्कर कार्य निरीक्षण प्रक्रिया तयार करू शकता. आम्ही झेपेलिनमध्ये नोटपॅड वापरतो, जिथे आम्ही कार्यांची स्थिती पाहतो:

हे स्वतः Airflow चे वेब इंटरफेस देखील असू शकते:

एअरफ्लो कोड ओपन सोर्स आहे, म्हणून आम्ही टेलीग्राममध्ये अॅलर्टिंग जोडले आहे. कार्याचे प्रत्येक चालू उदाहरण, एखादी त्रुटी आढळल्यास, टेलीग्राममधील गटाला स्पॅम करते, जिथे संपूर्ण विकास आणि समर्थन संघ असतो.

आम्हाला टेलीग्राम (आवश्यक असल्यास) द्वारे त्वरित प्रतिसाद मिळतो आणि झेपेलिनद्वारे आम्हाला एअरफ्लोमधील कार्यांचे एकंदर चित्र प्राप्त होते.

एकूण

एअरफ्लो हा मुख्यतः मुक्त स्रोत आहे आणि तुम्ही त्यातून चमत्काराची अपेक्षा करू नये. कार्य करणारे उपाय तयार करण्यासाठी वेळ आणि प्रयत्न करण्यास तयार रहा. ध्येय साध्य करण्यायोग्य आहे, माझ्यावर विश्वास ठेवा, ते योग्य आहे. विकासाची गती, लवचिकता, नवीन प्रक्रिया जोडण्याची सुलभता - तुम्हाला ते आवडेल. नक्कीच, आपल्याला प्रकल्पाच्या संस्थेकडे, एअरफ्लोची स्थिरता यावर खूप लक्ष देणे आवश्यक आहे: चमत्कार घडत नाहीत.

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

भविष्यासाठी योजना

प्रक्रियांची संख्या अपरिहार्यपणे वाढत आहे आणि एअरफ्लो इन्फ्रास्ट्रक्चरच्या बाबतीत आम्ही मुख्य गोष्ट करणार आहोत ते स्केलिंग आहे. आम्हाला एक एअरफ्लो क्लस्टर तयार करायचा आहे, सेलेरी कामगारांसाठी पायांची एक जोडी वाटप करायची आहे आणि जॉब शेड्यूलिंग प्रक्रिया आणि भांडारासह सेल्फ-डुप्लिकेट हेड बनवायचे आहे.

एपिलोग

हे, अर्थातच, मी एअरफ्लोबद्दल सांगू इच्छित सर्वकाही नाही, परंतु मी मुख्य मुद्दे हायलाइट करण्याचा प्रयत्न केला. भूक खाण्याने लागते, करून पहा तुम्हाला आवडेल :)

स्त्रोत: www.habr.com

बॅच डेटा प्रोसेसिंग प्रक्रिया सोयीस्करपणे आणि त्वरीत विकसित आणि देखरेख करण्यासाठी एअरफ्लो एक साधन आहे