By ड्रू न्यूबेरी, Gretel.ai . में सॉफ्टवेयर इंजीनियर

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

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

वायु प्रवाह क्या है

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

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

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

ग्रेटेल कैसे फिट बैठता है?

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

इन क्षमताओं को प्रदर्शित करने के लिए, हम एक ईटीएल पाइपलाइन का निर्माण करेंगे जो डेटाबेस से उपयोगकर्ता गतिविधि सुविधाओं को निकालती है, डेटासेट का सिंथेटिक संस्करण उत्पन्न करती है, और डेटासेट को S3 में सहेजती है। S3 में सहेजे गए सिंथेटिक डेटासेट के साथ, इसे डेटा वैज्ञानिकों द्वारा डाउनस्ट्रीम मॉडलिंग या विश्लेषण के लिए ग्राहक गोपनीयता से समझौता किए बिना उपयोग किया जा सकता है।

चीजों को शुरू करने के लिए, आइए पहले पाइपलाइन का एक विहंगम दृश्य देखें। इस आरेख में प्रत्येक नोड एक पाइपलाइन चरण, या वायु प्रवाह के संदर्भ में "कार्य" का प्रतिनिधित्व करता है।

उदाहरण एयरफ्लो पर ग्रेटेल सिंथेटिक्स पाइपलाइन।

हम पाइपलाइन को 3 चरणों में तोड़ सकते हैं, जैसा कि आप ईटीएल पाइपलाइन में पा सकते हैं:

• उद्धरण - Extract_features कार्य एक डेटाबेस को क्वेरी करेगा, और डेटा को सुविधाओं के एक सेट में बदल देगा जिसका उपयोग डेटा वैज्ञानिकों द्वारा मॉडल बनाने के लिए किया जा सकता है।
• सिंथेसाइज़ - generate_synthetic_features इनपुट के रूप में निकाली गई सुविधाओं को लेगा, सिंथेटिक मॉडल को प्रशिक्षित करेगा, और फिर ग्रेटेल एपीआई और क्लाउड सेवाओं का उपयोग करके सुविधाओं का सिंथेटिक सेट तैयार करेगा।
• ‍भार - upload_synthetic_features सुविधाओं के सिंथेटिक सेट को S3 में सहेजता है जहां इसे किसी भी डाउनस्ट्रीम मॉडल या विश्लेषण में डाला जा सकता है।

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

इसके अतिरिक्त, प्रत्येक घटक को विच्छेदित करने से पहले, एयरफ्लो पाइपलाइन को उसकी संपूर्णता में देखना सहायक हो सकता है, dags/airbnb_user_bookings.py. निम्नलिखित अनुभागों में कोड स्निपेट लिंक किए गए उपयोगकर्ता बुकिंग पाइपलाइन से निकाले जाते हैं।

विशेषताएं निकालें

 
 
पहला कार्य, Extract_features स्रोत डेटाबेस से कच्चे डेटा को निकालने और इसे सुविधाओं के एक सेट में बदलने के लिए जिम्मेदार है। यह एक आम है इंजीनियरिंग की सुविधा समस्या आपको किसी भी मशीन लर्निंग या एनालिटिक्स पाइपलाइन में मिल सकती है।

हमारे उदाहरण पाइपलाइन में हम एक पोस्टग्रेएसक्यूएल डेटाबेस का प्रावधान करेंगे और इसे एक से बुकिंग डेटा के साथ लोड करेंगे Airbnb कागल प्रतियोगिता.

इस डेटासेट में दो टेबल, उपयोगकर्ता और सत्र हैं। सत्र में एक विदेशी कुंजी संदर्भ, user_id होता है। इस संबंध का उपयोग करते हुए, हम उपयोगकर्ता द्वारा एकत्रित विभिन्न बुकिंग मीट्रिक वाली सुविधाओं का एक सेट तैयार करेंगे। निम्नलिखित आंकड़ा सुविधाओं के निर्माण के लिए उपयोग की जाने वाली SQL क्वेरी का प्रतिनिधित्व करता है।

session_features_by_user AS के साथ (user_id चुनें, गिनती करें (*) AS number_of_actions_taken, count(DISTINCT action_type) AS number_of_unique_actions, Round(avg(secs_elapsed)) AS avg_session_time_seconds, Round(max(secs_elapsed) AS_time_seconds, Round(max(secs_elapsed) AS_time_सेकेंड (अधिकतम) min_session_time_seconds, ( सेशन से गिनती (*) चुनें जहां s.user_id = user_id और s.action_type = 'booking_request') एएस टोटल_बुकिंग फ्रॉम सेशन ग्रुप बाय यूजर_आईडी ) यूज़र आईडी, यू.जेंडर, यू.एज, यू चुनें। .भाषा, u.signup_method, u.date_account_created, s.number_of_actions_taken, s.number_of_unique_actions, s.avg_session_time_seconds, s.min_session_time_seconds, s.max_session_time_सेकेंड।

SQL क्वेरी को तब हमारे Airflow पाइपलाइन से निष्पादित किया जाता है और निम्न कार्य परिभाषा का उपयोग करके एक मध्यवर्ती S3 स्थान पर लिखा जाता है।

@task() def Extract_features(sql_file: str) -> str: संदर्भ = get_current_context() sql_query = Path(sql_file).read_text() key = f"{context['dag_run'].run_id}_booking_features.csv" NamedTemporaryFile के साथ (mode="r+", suffix=".csv") tmp_csv के रूप में: postgres.copy_expert(f"copy ({sql_query}) to stdout with csv हेडर", tmp_csv.name ) s3.load_file( filename=tmp_csv.name, कुंजी = कुंजी, ) वापसी कुंजी

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

नीचे दिया गया स्क्रीनशॉट ऊपर से निष्कर्षण क्वेरी का एक नमूना परिणाम सेट दिखाता है। हम अगले भाग में इस डेटासेट का सिंथेटिक संस्करण बनाने का तरीका बताएंगे।

क्वेरी परिणाम पूर्वावलोकन।

Gretel API का उपयोग करके सुविधाओं का संश्लेषण करें

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

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

हुक से।ग्रेटेल आयात ग्रेटेलहुक ग्रेटेल = ग्रेटेलहुक () प्रोजेक्ट = ग्रेटेल.गेट_प्रोजेक्ट ()

एयरफ्लो कनेक्शन को कॉन्फ़िगर करने के तरीके के बारे में अधिक जानकारी के लिए, कृपया हमारे जीथब रिपोजिटरी को देखें README.

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

बुकिंग पाइपलाइन का जिक्र करते हुए, अब हम Generate_synthetic_features कार्य की समीक्षा करेंगे। यह चरण पिछले कार्य में निकाली गई सुविधाओं का उपयोग करके सिंथेटिक मॉडल को प्रशिक्षित करने के लिए जिम्मेदार है।

@task() def generate_synthetic_features(data_source: str) -> str: project = Gretel.get_project() model = project.create_model_obj(model_config="synthetics/default", data_source=s3.download_file(data_source) model.submit_cloud() मतदान (मॉडल) वापसी model.get_artifact_link("data_preview")

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

Project.create_model_obj का उपयोग करके मॉडल बनाते समय, model_config param मॉडल बनाने के लिए उपयोग किए जाने वाले सिंथेटिक मॉडल कॉन्फ़िगरेशन का प्रतिनिधित्व करता है। इस पाइपलाइन में, हम अपने का उपयोग कर रहे हैं डिफ़ॉल्ट मॉडल कॉन्फ़िगरेशन, लेकिन कई अन्य विन्यास विकल्प के उपलब्ध हैं.

मॉडल को कॉन्फ़िगर करने के बाद, हम model.submit_cloud() को कॉल करते हैं। यह प्रशिक्षण के लिए मॉडल प्रस्तुत करेगा और ग्रेटेल क्लाउड का उपयोग करके रिकॉर्ड निर्माण करेगा। कॉलिंग पोल (मॉडल) तब तक कार्य को रोक देगा जब तक कि मॉडल प्रशिक्षण पूरा नहीं कर लेता।

अब जब मॉडल को प्रशिक्षित किया गया है, तो हम उत्पन्न सिंथेटिक सुविधाओं को डाउनलोड करने के लिए एक लिंक वापस करने के लिए get_artifact_link का उपयोग करेंगे।

सुविधाओं के सिंथेटिक सेट का डेटा पूर्वावलोकन।

इस आर्टिफैक्ट लिंक का उपयोग अंतिम upload_synthetic_features चरण के इनपुट के रूप में किया जाएगा।

लोड सिंथेटिक विशेषताएं

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

@task() def upload_synthetic_features(data_set: str): संदर्भ = get_current_context() ओपन (data_set, "rb") के साथ synth_features के रूप में: s3.load_file_obj (file_obj=synth_features, key=f"{..._booking_features_synthetic.csv", )

यह कार्य काफी सीधा है। data_set इनपुट मान में Gretel के API से सिंथेटिक डेटासेट डाउनलोड करने के लिए एक हस्ताक्षरित HTTP लिंक होता है। कार्य उस फ़ाइल को एयरफ्लो कार्यकर्ता में पढ़ेगा, और फिर सिंथेटिक फीचर फ़ाइल को S3 बाल्टी में अपलोड करने के लिए पहले से कॉन्फ़िगर किए गए S3 हुक का उपयोग करेगा जहां डाउनस्ट्रीम उपभोक्ता या मॉडल इसे एक्सेस कर सकते हैं।

पाइपलाइन को व्यवस्थित करना

 
 
पिछले तीन खंडों में हमने एक डेटासेट को निकालने, संश्लेषित करने और लोड करने के लिए आवश्यक सभी कोड का अध्ययन किया है। अंतिम चरण इन कार्यों में से प्रत्येक को एक एयरफ्लो पाइपलाइन में एक साथ जोड़ना है।

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

feature_path = Extract_features ("/opt/airflow/dags/sql/session_rollups__by_user.sql") सिंथेटिक_डेटा = Generate_synthetic_features(feature_path) upload_synthetic_features(synthetic_data)

यदि आप इन मेथड कॉल्स के पथ का अनुसरण करते हैं, तो आपको अंततः एक ग्राफ मिलेगा जो हमारी मूल फीचर पाइपलाइन जैसा दिखता है।

एयरफ्लो पर ग्रेटेल सिंथेटिक्स पाइपलाइन।

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

चीजों को रैपिंग करना

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

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

फीचर पाइपलाइन के बारे में अधिक सारगर्भित शब्दों में सोचते हुए, अब हमारे पास एक ऐसा पैटर्न है जिसे किसी भी संख्या में नए SQL प्रश्नों के लिए फिर से तैयार किया जा सकता है। पाइपलाइन के एक नए संस्करण को तैनात करके, और प्रारंभिक SQL क्वेरी को स्वैप करके, हम किसी भी संभावित संवेदनशील क्वेरी को सिंथेटिक डेटासेट के साथ सामने रख सकते हैं जो ग्राहक की गोपनीयता को बनाए रखता है। कोड की एकमात्र पंक्ति जिसे बदलने की आवश्यकता है वह sql फ़ाइल का पथ है। कोई जटिल डेटा इंजीनियरिंग की आवश्यकता नहीं है।

पढ़ने के लिए धन्यवाद

 
 
हमें एक ईमेल भेजें hi@gretel.ai या हमसे जुड़ें सुस्त यदि आपके कोई प्रश्न या टिप्पणी हैं। हमें यह जानकर खुशी होगी कि आप एयरफ्लो का उपयोग कैसे कर रहे हैं और हम आपकी मौजूदा डेटा पाइपलाइनों के साथ सर्वोत्तम तरीके से कैसे एकीकृत हो सकते हैं।

 
जैव: ड्रू न्यूबेरी Gretel.ai में सॉफ्टवेयर इंजीनियर हैं।

मूल। अनुमति के साथ पुनर्प्रकाशित।

संबंधित:

• प्रीफेक्ट: पायथन के साथ अपनी पहली ईटीएल पाइपलाइन कैसे लिखें और शेड्यूल करें
• आपके डेटा साइंस पाइपलाइन को ऑप्टिमाइज़ करने के लिए 15 पायथन स्निपेट्स
• अपने पंडों के डेटाफ़्रेम को कैसे क्वेरी करें