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

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

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

Google क्लाउड की स्थापना

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

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

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

चित्रा 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 और फ़ंक्शन को कॉल करें publish с topic_path और डेटा. कृपया ध्यान दें कि हम आयात करते हैं generate_log_line हमारी स्क्रिप्ट से stream_logs, इसलिए सुनिश्चित करें कि ये फ़ाइलें एक ही फ़ोल्डर में हैं, अन्यथा आपको एक आयात त्रुटि मिलेगी। फिर हम इसे अपने 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, जो डेटा को स्कैन करता है और फ़ंक्शन का उपयोग करके पैटर्न सूची के आधार पर संबंधित पंक्ति को पुनर्प्राप्त करता है re.search. फ़ंक्शन अल्पविराम से अलग की गई स्ट्रिंग लौटाता है। यदि आप रेगुलर एक्सप्रेशन विशेषज्ञ नहीं हैं, तो मैं इसे जांचने की सलाह देता हूं ट्यूटोरियल और कोड की जांच करने के लिए नोटपैड में अभ्यास करें। इसके बाद हम एक कस्टम ParDo फ़ंक्शन को परिभाषित करते हैं जिसे कहा जाता है विभाजित करें, जो समानांतर प्रसंस्करण के लिए बीम परिवर्तन का एक रूप है। पायथन में, यह एक विशेष तरीके से किया जाता है - हमें एक ऐसा वर्ग बनाना होगा जो DoFn बीम वर्ग से प्राप्त हो। स्प्लिट फ़ंक्शन पिछले फ़ंक्शन से पार्स की गई पंक्ति लेता है और हमारी BigQuery तालिका में कॉलम नामों के अनुरूप कुंजी वाले शब्दकोशों की एक सूची लौटाता है। इस फ़ंक्शन के बारे में ध्यान देने योग्य बात यह है: मुझे आयात करना पड़ा datetime किसी फ़ंक्शन के अंदर इसे कार्यान्वित करने के लिए। मुझे फ़ाइल की शुरुआत में एक आयात त्रुटि मिल रही थी, जो अजीब थी। फिर यह सूची फ़ंक्शन को भेज दी जाती है WriteToBigQuery, जो बस हमारे डेटा को तालिका में जोड़ता है। बैच डेटाफ़्लो जॉब और स्ट्रीमिंग डेटाफ़्लो जॉब का कोड नीचे दिया गया है। बैच और स्ट्रीमिंग कोड के बीच एकमात्र अंतर यह है कि बैच में हम सीएसवी को पढ़ते हैं src_pathफ़ंक्शन का उपयोग करना ReadFromText बीम से.

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

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'"[A-Z][a-z]+', r'"(http|https)://[a-z]+.[a-z]+.[a-z]+']
    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'"[A-Z][a-z]+', r'"(http|https)://[a-z]+.[a-z]+.[a-z]+']
    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()

कन्वेयर शुरू करना

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

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

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

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

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

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

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

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

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

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

चित्र 5: बिगक्वेरी

निष्कर्ष

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

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

https://github.com/DFoly/User_log_pipeline

बस इतना ही। भाग एक पढ़ें.

स्रोत: www.habr.com