🥇Airflow सुविधाजनक र द्रुत रूपमा ब्याच डाटा प्रोसेसिंग प्रक्रियाहरू विकास र समर्थन गर्ने उपकरण हो। प्रोहोस्टर

नमस्ते, Habr! यस लेखमा म ब्याच डाटा प्रोसेसिङ प्रक्रियाहरू विकास गर्नका लागि एउटा उत्कृष्ट उपकरणको बारेमा कुरा गर्न चाहन्छु, उदाहरणका लागि, कर्पोरेट DWH वा तपाईंको DataLake को पूर्वाधारमा। हामी Apache Airflow को बारेमा कुरा गर्नेछौं (यसलाई पछि Airflow भनिन्छ)। यो अनुचित रूपमा Habré मा ध्यानबाट वञ्चित छ, र मुख्य भागमा म तपाईंलाई विश्वस्त गराउने प्रयास गर्नेछु कि कम्तिमा एयरफ्लो तपाईको ETL/ELT प्रक्रियाहरूको लागि अनुसूचक छनौट गर्दा हेर्न लायक छ।

पहिले, मैले टिङ्कफ बैंकमा काम गर्दा DWH को विषयमा लेखहरूको श्रृंखला लेखेको थिएँ। अब म Mail.Ru समूह टोलीको हिस्सा बनेको छु र गेमिङ क्षेत्रमा डेटा विश्लेषणको लागि प्लेटफर्म विकास गर्दैछु। वास्तवमा, समाचार र चाखलाग्दो समाधानहरू देखा पर्दा, मेरो टोली र म यहाँ डाटा एनालिटिक्सको लागि हाम्रो प्लेटफर्मको बारेमा कुरा गर्नेछौं।

प्रस्तावना

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

अब Airflow को मुख्य संस्थाहरु लाई हेरौं। तिनीहरूको सार र उद्देश्य बुझेर, तपाईंले आफ्नो प्रक्रिया वास्तुकलालाई उत्तम रूपमा व्यवस्थित गर्न सक्नुहुन्छ। सायद मुख्य इकाई निर्देशित एसाइक्लिक ग्राफ हो (यसपछि DAG भनिन्छ)।

DAG

एक DAG तपाइँको कार्यहरु को केहि अर्थपूर्ण संघ हो जुन तपाइँ एक विशेष तालिका अनुसार एक सख्त परिभाषित अनुक्रम मा पूरा गर्न चाहानुहुन्छ। एयरफ्लोले DAGs र अन्य निकायहरूसँग काम गर्नको लागि सुविधाजनक वेब इन्टरफेस प्रदान गर्दछ:

DAG यस्तो देखिन सक्छ:

विकासकर्ताले DAG डिजाइन गर्दा, DAG भित्रका कार्यहरू निर्माण गरिने अपरेटरहरूको सेट राख्छ। यहाँ हामी अर्को महत्त्वपूर्ण संस्थामा आउँछौं: वायुप्रवाह अपरेटर।

अपरेटरहरू

एक अपरेटर एक संस्था हो जसको आधारमा कार्य उदाहरणहरू सिर्जना गरिन्छ, जसले कार्य उदाहरणको कार्यान्वयनको क्रममा के हुनेछ भनेर वर्णन गर्दछ। GitHub बाट एयरफ्लो रिलीज पहिले नै प्रयोग गर्न तयार अपरेटरहरूको सेट समावेश गर्दछ। उदाहरणहरू:

BashOperator - bash आदेश कार्यान्वयन गर्न को लागी अपरेटर।
PythonOperator - पाइथन कोड कल गर्नका लागि अपरेटर।
इमेल अपरेटर - इमेल पठाउनको लागि अपरेटर।
HTTPOperator - http अनुरोधहरूसँग काम गर्ने अपरेटर।
SqlOperator - SQL कोड कार्यान्वयन गर्न को लागी अपरेटर।
सेन्सर घटनाको लागि पर्खने अपरेटर हो (आवश्यक समयको आगमन, आवश्यक फाइलको उपस्थिति, डाटाबेसमा लाइन, API बाट प्रतिक्रिया, आदि।)।

त्यहाँ थप विशिष्ट अपरेटरहरू छन्: DockerOperator, HiveOperator, S3FileTransferOperator, PrestoToMysqlOperator, SlackOperator।

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

अर्को, कार्यहरूको यी सबै उदाहरणहरू कार्यान्वयन गर्न आवश्यक छ, र अब हामी अनुसूचकको बारेमा कुरा गर्नेछौं।

अनुसूचीकर्ता

Airflow को कार्य शेड्यूलर मा निर्मित छ अजवाइन। सेलेरी एक पाइथन पुस्तकालय हो जसले तपाईंलाई कार्यहरूको एसिन्क्रोनस र वितरण गरिएको कार्यान्वयनलाई लाम व्यवस्थित गर्न अनुमति दिन्छ। एयरफ्लो पक्षमा, सबै कार्यहरू पूलहरूमा विभाजित छन्। पोखरीहरू म्यानुअल रूपमा सिर्जना गरिएका छन्। सामान्यतया, तिनीहरूको उद्देश्य स्रोतसँग काम गर्ने कार्यभारलाई सीमित गर्नु वा 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 स्तरमा परिभाषित पूल कार्य स्तरमा ओभरराइड गर्न सकिन्छ।
एउटा छुट्टै प्रक्रिया, शेड्युलर, एयरफ्लोमा भएका सबै कार्यहरूको तालिका बनाउन जिम्मेवार हुन्छ। वास्तवमा, शेड्युलरले कार्यान्वयनको लागि कार्यहरू सेट गर्ने सबै मेकानिकहरूसँग सम्झौता गर्दछ। कार्य कार्यान्वयन हुनु अघि धेरै चरणहरू मार्फत जान्छ:

डीएजीमा पहिलेका कामहरू पूरा भइसकेका छन्, नयाँ लाइनमा राख्न सकिन्छ।
कार्यहरूको प्राथमिकताको आधारमा लाम क्रमबद्ध गरिएको छ (प्राथमिकताहरू पनि नियन्त्रण गर्न सकिन्छ), र यदि पूलमा नि: शुल्क स्लट छ भने, कार्य सञ्चालनमा लिन सकिन्छ।
यदि त्यहाँ एक नि: शुल्क कार्यकर्ता सेलेरी छ भने, कार्य यसलाई पठाइन्छ; तपाईंले समस्यामा प्रोग्राम गर्नुभएको काम एक वा अर्को अपरेटर प्रयोग गरेर सुरु हुन्छ।

पर्याप्त सरल।

शेड्युलर सबै DAGs र DAG भित्रका सबै कार्यहरूको सेटमा चल्छ।

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 को कार्यान्वयन पाइथनमा कोड हो, त्यसैले हामीसँग काम गर्दा कोडको मात्रा घटाउने धेरै सुविधाजनक तरिका छ, उदाहरणका लागि, छायादार स्रोतहरूसँग। मानौं तपाईंसँग स्रोतको रूपमा तीन 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)

DAG यस्तो देखिन्छ:

यस अवस्थामा, तपाइँ सेटिङहरू समायोजन गरेर र DAG अद्यावधिक गरेर शार्ड थप्न वा हटाउन सक्नुहुन्छ। आरामदायी!

तपाईले थप जटिल कोड उत्पादन पनि प्रयोग गर्न सक्नुहुन्छ, उदाहरणका लागि, डाटाबेसको रूपमा स्रोतहरूसँग काम गर्नुहोस् वा तालिका संरचना वर्णन गर्नुहोस्, तालिकासँग काम गर्ने एल्गोरिदम, र DWH पूर्वाधारका सुविधाहरूलाई ध्यानमा राखेर, प्रक्रिया उत्पन्न गर्नुहोस्। तपाईंको भण्डारणमा N तालिकाहरू लोड गर्नका लागि। वा, उदाहरणका लागि, एपीआईसँग काम गर्ने जसले सूचीको रूपमा प्यारामिटरसँग काम गर्न समर्थन गर्दैन, तपाइँ यस सूचीबाट DAG मा N कार्यहरू उत्पन्न गर्न सक्नुहुन्छ, एपीआईमा अनुरोधहरूको समानान्तरलाई पूलमा सीमित गर्नुहोस् र स्क्र्याप गर्नुहोस्। API बाट आवश्यक डाटा। लचिलो!

भण्डार

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

निगरानी गर्दै

भण्डारको सरलतालाई ध्यानमा राख्दै, तपाइँ तपाइँको लागि सुविधाजनक कार्य अनुगमन प्रक्रिया निर्माण गर्न सक्नुहुन्छ। हामी Zeppelin मा एक नोटप्याड प्रयोग गर्छौं, जहाँ हामी कार्यहरूको स्थिति हेर्छौं:

यो पनि Airflow को वेब इन्टरफेस हुन सक्छ:

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

हामीले टेलिग्राम (यदि आवश्यक भएमा) मार्फत द्रुत प्रतिक्रिया प्राप्त गर्छौं, र Zeppelin मार्फत हामीले Airflow मा कार्यहरूको समग्र तस्वीर प्राप्त गर्छौं।

कुल

एयरफ्लो मुख्य रूपमा खुला स्रोत हो, र तपाईंले यसबाट चमत्कारहरूको आशा गर्नु हुँदैन। काम गर्ने समाधान निर्माण गर्न समय र प्रयासमा राख्न तयार हुनुहोस्। लक्ष्य प्राप्त गर्न सकिन्छ, मलाई विश्वास गर्नुहोस्, यो यसको लायक छ। विकासको गति, लचिलोपन, नयाँ प्रक्रियाहरू थप्न सजिलो - तपाईंलाई यो मनपर्नेछ। निस्सन्देह, तपाईंले परियोजनाको संगठन, एयरफ्लोको स्थिरतामा धेरै ध्यान दिन आवश्यक छ: चमत्कारहरू हुँदैन।

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

भविष्यको लागि योजनाहरू

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

Epilogue

यो, निस्सन्देह, मैले एयरफ्लोको बारेमा बताउन चाहने सबै कुरा होइन, तर मैले मुख्य बुँदाहरू हाइलाइट गर्ने प्रयास गरें। भोक खाएसँगै आउँछ, प्रयास गर्नुहोस् र तपाईलाई यो मनपर्नेछ :)

स्रोत: www.habr.com

एयरफ्लो ब्याच डाटा प्रोसेसिंग प्रक्रियाहरू सहज र छिटो विकास र कायम राख्नको लागि एक उपकरण हो