डेटा प्रोसेसिंग स्ट्रीमिंग के लिए एक पाइपलाइन बनाएं। भाग 1

सभी को नमस्कार। दोस्तों, हम आपके साथ विशेष रूप से डाटा इंजीनियर पाठ्यक्रम के छात्रों के लिए तैयार किए गए एक लेख का अनुवाद साझा कर रहे हैं। चलो चलते हैं!

अपाचे बीम और वास्तविक समय पाइपलाइनों के लिए डेटाफ्लो

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

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

इसलिए क्लाउड डेटाबेस में डेटा डालना मेरी पहली समस्या को हल करने के लिए एक स्मार्ट तरीका की तरह लग रहा था, लेकिन मैं समस्या नंबर 2 के साथ क्या कर सकता था? सौभाग्य से, इस डेटा को ऐसे वातावरण में स्थानांतरित करने का एक तरीका था जहां मैं पायथन और Google क्लाउड प्लेटफ़ॉर्म (GCP) जैसे उपकरणों का उपयोग कर सकता था। हालांकि, यह एक लंबी प्रक्रिया थी, इसलिए मुझे कुछ ऐसा करने की जरूरत थी, जिससे मुझे डेटा ट्रांसफर के अंत का इंतजार करते हुए विकास जारी रखने की अनुमति मिले। इसका हल मैं पायथन में फ़ेकर लाइब्रेरी का उपयोग करके नकली डेटा बनाना था। मैंने पहले कभी इस पुस्तकालय का उपयोग नहीं किया था, लेकिन जल्दी से एहसास हुआ कि यह कितना उपयोगी था। इस दृष्टिकोण का उपयोग करने से मुझे वास्तविक डेटा के बिना कोड लिखना और पाइपलाइन का परीक्षण शुरू करने की अनुमति मिली।

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

जीसीपी और अपाचे बीम का परिचय

Google क्लाउड प्लेटफ़ॉर्म बड़े डेटा को संसाधित करने के लिए वास्तव में उपयोगी उपकरणों का एक सेट प्रदान करता है। यहां कुछ उपकरण दिए गए हैं जिनका मैं उपयोग करूंगा:

• पब / सब प्रकाशक-सब्सक्राइबर टेम्पलेट का उपयोग करके एक संदेश सेवा है जो हमें वास्तविक समय में डेटा प्राप्त करने की अनुमति देता है।
• डेटाफ्लो एक ऐसी सेवा है जो डेटा पाइपलाइनों के निर्माण को सरल बनाती है और बुनियादी ढांचे को स्केल करने जैसे कार्यों को स्वचालित रूप से हल करती है, जिसका अर्थ है कि हम केवल अपनी पाइपलाइन के लिए कोड लिखने पर ध्यान केंद्रित कर सकते हैं।
• BigQuery क्लाउड-आधारित डेटा वेयरहाउस है। यदि आप अन्य SQL डेटाबेस से परिचित हैं, तो आपको BigQuery के साथ लंबे समय तक व्यवहार नहीं करना होगा।
• और अंत में, हम अपाचे बीम का उपयोग करेंगे, अर्थात्, हमारी पाइपलाइन बनाने के लिए पायथन संस्करण पर ध्यान केंद्रित करें। यह उपकरण हमें स्ट्रीमिंग या बैच प्रसंस्करण के लिए एक पाइपलाइन बनाने की अनुमति देगा जो जीसीपी के साथ एकीकृत होता है। यह समानांतर प्रसंस्करण के लिए विशेष रूप से उपयोगी है और निष्कर्षण, परिवर्तन, और लोडिंग (ETL) जैसे कार्यों के लिए उपयुक्त है, इसलिए यदि हमें परिवर्तनों या गणनाओं के साथ एक स्थान से दूसरे स्थान पर डेटा स्थानांतरित करने की आवश्यकता है, तो बीम एक अच्छा विकल्प है।

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

हमारे कन्वेयर का विज़ुअलाइज़ेशन

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

1. हमारे उपयोगकर्ताओं का लॉग डेटा पब / सब सेक्शन में प्रकाशित होता है।
2. हम पब / सब से जुड़ेंगे और पायथन और बीम (चित्रा 1 में चरण 3 और 4) का उपयोग करके डेटा को उपयुक्त प्रारूप में परिवर्तित करेंगे।
3. डेटा को परिवर्तित करने के बाद, बीम तब BigQuery से कनेक्ट होगा और उन्हें हमारी तालिका में जोड़ देगा (चित्रा 1 में चरण 4 और 5)।
4. विश्लेषण के लिए, हम विभिन्न उपकरणों जैसे कि झांकी और पायथन का उपयोग करके बिगविक से जुड़ सकते हैं।

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


चित्र 1: मुख्य डेटा पाइपलाइन

Faker का उपयोग करके छद्म डेटा बनाना

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

192.52.197.161 - - [30/Apr/2019:21:11:42] "PUT /tag/category/tag HTTP/1.1" [401] 155 "https://harris-lopez.com/categories/about/" "Mozilla/5.0 (Macintosh; PPC Mac OS X 10_11_2) AppleWebKit/5312 (KHTML, like Gecko) Chrome/34.0.855.0 Safari/5312"

उपरोक्त पंक्ति के आधार पर, हम नीचे दिए गए ब्रेसेस में 7 वेरिएबल्स का उपयोग करके अपना LINE वैरिएबल बनाना चाहते हैं। हम उन्हें थोड़ी देर बाद अपनी तालिका स्कीमा में चर नामों के रूप में भी उपयोग करेंगे।

LINE = """\
{remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"\
"""

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

fake.ipv4()
fake.uri_path()
fake.uri()
fake.user_agent()


 from faker import Faker import time import random import os import numpy as np from datetime import datetime, timedelta LINE = """\ {remote_addr} - - [{time_local}] "{request_type} {request_path} HTTP/1.1" [{status}] {body_bytes_sent} "{http_referer}" "{http_user_agent}"\ """ def generate_log_line(): fake = Faker() now = datetime.now() remote_addr = fake.ipv4() time_local = now.strftime('%d/%b/%Y:%H:%M:%S') request_type = random.choice(["GET", "POST", "PUT"]) request_path = "/" + fake.uri_path() status = np.random.choice([200, 401, 404], p = [0.9, 0.05, 0.05]) body_bytes_sent = random.choice(range(5, 1000, 1)) http_referer = fake.uri() http_user_agent = fake.user_agent() log_line = LINE.format( remote_addr=remote_addr, time_local=time_local, request_type=request_type, request_path=request_path, status=status, body_bytes_sent=body_bytes_sent, http_referer=http_referer, http_user_agent=http_user_agent ) return log_line 

पहले भाग का अंत।

आने वाले दिनों में, हम आपके साथ लेख को जारी रखेंगे, लेकिन अब हम परंपरागत रूप से टिप्पणियों का इंतजार कर रहे हैं;;

दूसरा भाग