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

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

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

Google क्लाउड सेटअप

नोट: मैंने पाइपलाइन शुरू करने और उपयोगकर्ता लॉग डेटा को प्रकाशित करने के लिए Google क्लाउड शेल का उपयोग किया, क्योंकि मुझे पायथन में पाइपलाइन चलाने में समस्याएं थीं। 3. Google क्लाउड शेल ने पायथन 2 का उपयोग किया है, जो अपाचे बीम के साथ बेहतर संगत है।

कन्वेयर शुरू करने के लिए, हमें सेटिंग्स में थोड़ा सीमांकन करने की आवश्यकता है। आपमें से जिन्होंने पहले GCP का उपयोग नहीं किया है, आपको इस पृष्ठ पर निम्नलिखित 6 चरणों को पूरा करना होगा।

उसके बाद, हमें अपनी स्क्रिप्ट्स को Google क्लाउड स्टोरेज पर अपलोड करना होगा और उन्हें हमारे Google क्लाउड शेल में कॉपी करना होगा। क्लाउड स्टोरेज पर अपलोड करना काफी तुच्छ है (एक विवरण यहां पाया जा सकता है )। अपनी फ़ाइलों को कॉपी करने के लिए, हम टूलबार से Google क्लाउड शेल को नीचे चित्र 2 में बाईं ओर पहले आइकन पर क्लिक करके खोल सकते हैं।


चित्र 2

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

# Copy file from cloud storage gsutil cp gs://<YOUR-BUCKET>/ * . sudo pip install apache-beam[gcp] oauth2client==3.0.0 sudo pip install -U pip sudo pip install Faker==1.0.2 # Environment variables BUCKET=<YOUR-BUCKET> PROJECT=<YOUR-PROJECT> 

हमारे डेटाबेस और तालिका बनाना

सभी कॉन्फ़िगरेशन चरणों को पूरा करने के बाद, अगली चीज़ जो हमें करने की ज़रूरत है वह है BigQuery में एक डेटासेट और टेबल बनाना। ऐसा करने के कई तरीके हैं, लेकिन सबसे आसान यह है कि पहले Google डेटा कंसोल का उपयोग करके एक डेटासेट बनाया जाए। स्कीमा के साथ तालिका बनाने के लिए आप निम्न लिंक में दिए गए चरणों का पालन कर सकते हैं। हमारी तालिका में प्रत्येक उपयोगकर्ता लॉग के घटकों के अनुरूप 7 कॉलम होंगे। सुविधा के लिए, हम सभी स्तंभों को स्ट्रिंग्स (प्रकार स्ट्रिंग) के रूप में परिभाषित करेंगे, जो समयबद्ध चर के अपवाद के साथ है, और उन्हें पहले बनाए गए चर के अनुसार नाम दें। हमारी तालिका का लेआउट चित्र 3 जैसा दिखना चाहिए।


चित्रा 3. टेबल लेआउट

उपयोगकर्ता लॉग डेटा प्रकाशित करें

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

नीचे दिया गया कोड ऊपर बताए गए लॉग डेटा को जेनरेट करने के लिए हमारी स्क्रिप्ट को कॉल करता है, और फिर लॉग / पब को जोड़ता है और भेजता है। केवल एक चीज जो हमें करने की आवश्यकता है, वह है एक प्रकाशक topic_path ऑब्जेक्ट, topic_path विधि का उपयोग करके विषय को पथ निर्दिष्ट करें और publish फ़ंक्शन को topic_path और डेटा के साथ कॉल करें। कृपया ध्यान दें कि हम अपनी stream_logs स्क्रिप्ट से stream_logs आयात generate_log_line , इसलिए सुनिश्चित करें कि ये फ़ाइलें उसी फ़ोल्डर में हैं, अन्यथा आपको एक आयात त्रुटि मिलेगी। तब हम इसका उपयोग करके अपने Google कंसोल के माध्यम से चला सकते हैं:

 python publish.py 

 from stream_logs import generate_log_line import logging from google.cloud import pubsub_v1 import random import time PROJECT_ID="user-logs-237110" TOPIC = "userlogs" publisher = pubsub_v1.PublisherClient() topic_path = publisher.topic_path(PROJECT_ID, TOPIC) def publish(publisher, topic, message): data = message.encode('utf-8') return publisher.publish(topic_path, data = data) def callback(message_future): # When timeout is unspecified, the exception method waits indefinitely. if message_future.exception(timeout=30): print('Publishing message on {} threw an Exception {}.'.format( topic_name, message_future.exception())) else: print(message_future.result()) if __name__ == '__main__': while True: line = generate_log_line() print(line) message_future = publish(publisher, topic_path, line) message_future.add_done_callback(callback) sleep_time = random.choice(range(1, 3, 1)) time.sleep(sleep_time) 

जैसे ही फ़ाइल शुरू होती है, हम कंसोल में लॉग डेटा के आउटपुट का निरीक्षण कर सकते हैं, जैसा कि नीचे दिए गए चित्र में दिखाया गया है। यह स्क्रिप्ट तब तक काम करेगी जब तक हम इसे पूरा करने के लिए CTRL + C का उपयोग नहीं करते।


चित्रा 4. publish_logs.py आउटपुट

हमारी पाइपलाइन के लिए कोड लिखना

अब जब हमने सब कुछ तैयार कर लिया है, तो हम सबसे दिलचस्प भाग के लिए आगे बढ़ सकते हैं - बीम और पायथन का उपयोग करके अपनी पाइपलाइन का कोड लिखना। बीम पाइपलाइन बनाने के लिए, हमें एक पाइपलाइन ऑब्जेक्ट (पी) बनाने की आवश्यकता है। पाइप लाइन ऑब्जेक्ट बनाने के बाद, हम pipe (|) ऑपरेटर का उपयोग करके एक के बाद एक कई कार्य कर सकते हैं। सामान्य तौर पर, वर्कफ़्लो नीचे की छवि जैसा दिखता है।

 [Final Output PCollection] = ([Initial Input PCollection] | [First Transform] | [Second Transform] | [Third Transform]) 

हमारे कोड में, हम दो उपयोगकर्ता-परिभाषित फ़ंक्शन बनाएंगे। regex_clean फ़ंक्शन, जो डेटा को स्कैन करता है और पुन: खोज फ़ंक्शन का उपयोग करके PATTERNS सूची के आधार पर संबंधित पंक्ति को पुनर्प्राप्त करता है। फ़ंक्शन एक अल्पविराम-पृथक स्ट्रिंग देता है। यदि आप एक नियमित अभिव्यक्ति विशेषज्ञ नहीं हैं, तो मैं सुझाव देता हूं कि आप इस ट्यूटोरियल को पढ़ें और कोड की जांच करने के लिए नोटपैड में अभ्यास करें। उसके बाद, हम स्प्लिट नामक एक कस्टम ParDo फ़ंक्शन को परिभाषित करते हैं, जो समानांतर प्रसंस्करण के लिए बीम परिवर्तन का एक रूप है। पायथन में, यह एक विशेष तरीके से किया जाता है - हमें एक वर्ग बनाना होगा जो कि DoFn बीम वर्ग से विरासत में मिला है। स्प्लिट फ़ंक्शन पिछले फ़ंक्शन से एक पार्स किया गया स्ट्रिंग लेता है और हमारे BigQuery तालिका में कॉलम नामों के अनुरूप कुंजियों के साथ शब्दकोशों की एक सूची देता है। इस फ़ंक्शन के बारे में कुछ ध्यान देने योग्य बात है: मुझे इसे काम करने के लिए फ़ंक्शन के अंदर datetime आयात करना पड़ा। मुझे फ़ाइल की शुरुआत में एक आयात त्रुटि मिली, जो अजीब थी। यह सूची तब WriteToBigQuery फ़ंक्शन को दी जाती है, जो हमारे डेटा को केवल तालिका में जोड़ता है। बैच डेटाफ़्लो जॉब और स्ट्रीमिंग डेटाफ़्लो जॉब का कोड नीचे दिखाया गया है। बैच और स्ट्रीम कोड के बीच एकमात्र अंतर यह है कि बैच प्रोसेसिंग में हम बीम से ReadFromText फ़ंक्शन का उपयोग करके src_path से CSV पढ़ते हैं।

बैचफ्लो डाटा जॉब (पैकेट प्रोसेसिंग)

 import apache_beam as beam from apache_beam.options.pipeline_options import PipelineOptions from google.cloud import bigquery import re import logging import sys PROJECT='user-logs-237110' schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING' src_path = "user_log_fileC.txt" def regex_clean(data): PATTERNS = [r'(^\S+\.[\S+\.]+\S+)\s',r'(?<=\[).+?(?=\])', r'\"(\S+)\s(\S+)\s*(\S*)\"',r'\s(\d+)\s',r"(?<=\[).\d+(?=\])", r'\"[AZ][az]+', r'\"(http|https)://[az]+.[az]+.[az]+'] result = [] for match in PATTERNS: try: reg_match = re.search(match, data).group() if reg_match: result.append(reg_match) else: result.append(" ") except: print("There was an error with the regex search") result = [x.strip() for x in result] result = [x.replace('"', "") for x in result] res = ','.join(result) return res class Split(beam.DoFn): def process(self, element): from datetime import datetime element = element.split(",") d = datetime.strptime(element[1], "%d/%b/%Y:%H:%M:%S") date_string = d.strftime("%Y-%m-%d %H:%M:%S") return [{ 'remote_addr': element[0], 'timelocal': date_string, 'request_type': element[2], 'status': element[3], 'body_bytes_sent': element[4], 'http_referer': element[5], 'http_user_agent': element[6] }] def main(): p = beam.Pipeline(options=PipelineOptions()) (p | 'ReadData' >> beam.io.textio.ReadFromText(src_path) | "clean address" >> beam.Map(regex_clean) | 'ParseCSV' >> beam.ParDo(Split()) | 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema, write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND) ) p.run() if __name__ == '__main__': logger = logging.getLogger().setLevel(logging.INFO) main() 

डेटाफ़्लो जॉब स्ट्रीमिंग

 from apache_beam.options.pipeline_options import PipelineOptions from google.cloud import pubsub_v1 from google.cloud import bigquery import apache_beam as beam import logging import argparse import sys import re PROJECT="user-logs-237110" schema = 'remote_addr:STRING, timelocal:STRING, request_type:STRING, status:STRING, body_bytes_sent:STRING, http_referer:STRING, http_user_agent:STRING' TOPIC = "projects/user-logs-237110/topics/userlogs" def regex_clean(data): PATTERNS = [r'(^\S+\.[\S+\.]+\S+)\s',r'(?<=\[).+?(?=\])', r'\"(\S+)\s(\S+)\s*(\S*)\"',r'\s(\d+)\s',r"(?<=\[).\d+(?=\])", r'\"[AZ][az]+', r'\"(http|https)://[az]+.[az]+.[az]+'] result = [] for match in PATTERNS: try: reg_match = re.search(match, data).group() if reg_match: result.append(reg_match) else: result.append(" ") except: print("There was an error with the regex search") result = [x.strip() for x in result] result = [x.replace('"', "") for x in result] res = ','.join(result) return res class Split(beam.DoFn): def process(self, element): from datetime import datetime element = element.split(",") d = datetime.strptime(element[1], "%d/%b/%Y:%H:%M:%S") date_string = d.strftime("%Y-%m-%d %H:%M:%S") return [{ 'remote_addr': element[0], 'timelocal': date_string, 'request_type': element[2], 'body_bytes_sent': element[3], 'status': element[4], 'http_referer': element[5], 'http_user_agent': element[6] }] def main(argv=None): parser = argparse.ArgumentParser() parser.add_argument("--input_topic") parser.add_argument("--output") known_args = parser.parse_known_args(argv) p = beam.Pipeline(options=PipelineOptions()) (p | 'ReadData' >> beam.io.ReadFromPubSub(topic=TOPIC).with_output_types(bytes) | "Decode" >> beam.Map(lambda x: x.decode('utf-8')) | "Clean Data" >> beam.Map(regex_clean) | 'ParseCSV' >> beam.ParDo(Split()) | 'WriteToBigQuery' >> beam.io.WriteToBigQuery('{0}:userlogs.logdata'.format(PROJECT), schema=schema, write_disposition=beam.io.BigQueryDisposition.WRITE_APPEND) ) result = p.run() result.wait_until_finish() if __name__ == '__main__': logger = logging.getLogger().setLevel(logging.INFO) main() 

कन्वेयर शुरू

हम कई अलग-अलग तरीकों से पाइपलाइन शुरू कर सकते हैं। यदि हम चाहते थे, तो हम इसे टर्मिनल से स्थानीय रूप से चला सकते हैं, दूरस्थ रूप से जीसीपी में प्रवेश कर सकते हैं।

 python -m main_pipeline_stream.py \ --input_topic "projects/user-logs-237110/topics/userlogs" \ --streaming 

हालाँकि, हम इसे DataFlow का उपयोग करके लॉन्च करने जा रहे हैं। हम निम्न आवश्यक पैरामीटर सेट करके नीचे दिए गए कमांड का उपयोग कर सकते हैं।

• project - आपके GCP प्रोजेक्ट की आईडी।
• runner एक पाइपलाइन runner जो आपके कार्यक्रम का विश्लेषण करेगा और आपकी पाइपलाइन का निर्माण करेगा। क्लाउड में चलने के लिए, आपको एक DataflowRunner निर्दिष्ट करना होगा।
• staging_location - प्रक्रिया हैंडलर द्वारा आवश्यक कोड पैकेट को अनुक्रमित करने के लिए क्लाउड staging_location क्लाउड स्टोरेज का मार्ग।
• temp_location - पाइपलाइन के संचालन के दौरान बनाई गई अस्थायी नौकरी फ़ाइलों को होस्ट करने के लिए क्लाउड temp_location क्लाउड स्टोरेज का मार्ग।
• streaming

 python main_pipeline_stream.py \ --runner DataFlow \ --project $PROJECT \ --temp_location $BUCKET/tmp \ --staging_location $BUCKET/staging --streaming 

जब यह कमांड चल रही होती है, तो हम Google कंसोल में DataFlow टैब पर जा सकते हैं और अपनी पाइपलाइन देख सकते हैं। पाइपलाइन पर क्लिक करके, हमें चित्र 4 के समान कुछ देखना चाहिए। डीबगिंग उद्देश्यों के लिए, लॉग में जाने के लिए और फिर स्टैकड्राइवर के लिए विस्तृत लॉग देखने के लिए बहुत उपयोगी हो सकता है। इससे मुझे कई मामलों में पाइपलाइन के साथ समस्याओं को हल करने में मदद मिली।


चित्रा 4: बीम कन्वेयर

BigQuery में हमारे डेटा तक पहुँचें

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

 SELECT * FROM `user-logs-237110.userlogs.logdata` LIMIT 10; 

चित्र 5: BigQuery

निष्कर्ष

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

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

https://github.com/DFoly/User_log_pipeline


वह सब है। पहला भाग पढ़ें