🥇 स्ट्रीम डेटा प्रोसेसिंग पाइपलाइन तयार करणे. भाग १

सर्वांना नमस्कार. आम्ही लेखाच्या शेवटच्या भागाचे भाषांतर शेअर करत आहोत, जे विशेषतः अभ्यासक्रमाच्या विद्यार्थ्यांसाठी तयार केले आहे. डेटा अभियंता. पहिला भाग तुम्ही वाचू शकता येथे.

रिअल-टाइम पाइपलाइनसाठी अपाचे बीम आणि डेटाफ्लो

Google क्लाउड सेट करत आहे

टीप: मी पाइपलाइन चालवण्यासाठी आणि सानुकूल लॉग डेटा प्रकाशित करण्यासाठी Google क्लाउड शेलचा वापर केला कारण मला पायथन 3 मध्ये पाइपलाइन चालवताना समस्या येत होती. Google क्लाउड शेल पायथन 2 वापरते, जे Apache Beam शी अधिक सुसंगत आहे.

पाइपलाइन सुरू करण्यासाठी, आम्हाला सेटिंग्जमध्ये थोडेसे खोदणे आवश्यक आहे. तुमच्यापैकी ज्यांनी आधी GCP वापरला नाही त्यांच्यासाठी, तुम्हाला यामध्ये वर्णन केलेल्या खालील 6 पायऱ्या फॉलो कराव्या लागतील पृष्ठ.

यानंतर, आम्हाला आमच्या स्क्रिप्ट्स Google Cloud Storage वर अपलोड कराव्या लागतील आणि त्या आमच्या Google Cloud Shel वर कॉपी कराव्या लागतील. क्लाउड स्टोरेजवर अपलोड करणे अगदी क्षुल्लक आहे (वर्णन सापडू शकते येथे). आमच्या फायली कॉपी करण्यासाठी, आम्ही खालील आकृती 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. टेबल लेआउट

वापरकर्ता लॉग डेटा प्रकाशित करत आहे

Pub/Sub हा आमच्या पाइपलाइनचा एक महत्त्वाचा घटक आहे कारण ते एकाधिक स्वतंत्र अनुप्रयोगांना एकमेकांशी संवाद साधण्याची परवानगी देते. विशेषतः, ते एक मध्यस्थ म्हणून कार्य करते जे आम्हाला अनुप्रयोगांमध्ये संदेश पाठविण्यास आणि प्राप्त करण्यास अनुमती देते. प्रथम आपल्याला एक विषय तयार करणे आवश्यक आहे. कन्सोलमध्ये फक्त Pub/Sub वर जा आणि CREATE TOPIC वर क्लिक करा.

वर परिभाषित केलेला लॉग डेटा तयार करण्यासाठी खालील कोड आमच्या स्क्रिप्टला कॉल करतो आणि नंतर लॉग जोडतो आणि Pub/Sub ला पाठवतो. आपल्याला फक्त एक वस्तू तयार करण्याची आवश्यकता आहे PublisherClient, पद्धत वापरून विषयाचा मार्ग निर्दिष्ट करा 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 + सीते पूर्ण करण्यासाठी.

आकृती 4. आउटपुट publish_logs.py

आमचा पाइपलाइन कोड लिहित आहे

आता आम्ही सर्वकाही तयार केले आहे, आम्ही मजेदार भाग सुरू करू शकतो - बीम आणि पायथन वापरून आमच्या पाइपलाइनचे कोडिंग. बीम पाइपलाइन तयार करण्यासाठी, आम्हाला पाइपलाइन ऑब्जेक्ट (पी) तयार करणे आवश्यक आहे. एकदा आपण पाइपलाइन ऑब्जेक्ट तयार केल्यावर, आपण ऑपरेटर वापरून एकामागून एक अनेक फंक्शन्स लागू करू शकतो pipe (|). सर्वसाधारणपणे, वर्कफ्लो खालील प्रतिमेप्रमाणे दिसते.

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

आमच्या कोडमध्ये, आम्ही दोन कस्टम फंक्शन्स तयार करू. कार्य regex_clean, जे डेटा स्कॅन करते आणि फंक्शन वापरून PATTERNS सूचीवर आधारित संबंधित पंक्ती पुनर्प्राप्त करते re.search. फंक्शन स्वल्पविरामाने विभक्त केलेली स्ट्रिंग मिळवते. तुम्ही नियमित अभिव्यक्ती तज्ञ नसल्यास, मी हे तपासण्याची शिफारस करतो ट्यूटोरियल आणि कोड तपासण्यासाठी नोटपॅडमध्ये सराव करा. यानंतर आम्ही कस्टम ParDo फंक्शन परिभाषित करतो स्प्लिट, जी समांतर प्रक्रियेसाठी बीम ट्रान्सफॉर्मची भिन्नता आहे. पायथनमध्ये, हे एका विशेष प्रकारे केले जाते - आपण DoFn बीम वर्गाकडून वारसा मिळालेला वर्ग तयार केला पाहिजे. स्प्लिट फंक्शन मागील फंक्शनमधून पार्स केलेली पंक्ती घेते आणि आमच्या BigQuery टेबलमधील स्तंभ नावांशी संबंधित की सह शब्दकोशांची सूची मिळवते. या कार्याबद्दल लक्षात घेण्यासारखे काहीतरी आहे: मला आयात करावे लागले datetime कार्य करण्यासाठी फंक्शनच्या आत. फाईलच्या सुरुवातीला मला आयात त्रुटी प्राप्त होत होती, जी विचित्र होती. ही यादी नंतर फंक्शनला दिली जाते ToBigQuery लिहा, जे फक्त आमचा डेटा टेबलमध्ये जोडते. बॅच डेटाफ्लो जॉब आणि स्ट्रीमिंग डेटाफ्लो जॉबसाठी कोड खाली दिलेला आहे. बॅच आणि स्ट्रीमिंग कोडमधील फरक एवढाच आहे की बॅचमध्ये आपण CSV वरून वाचतो 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 — तुमच्या GCP प्रकल्पाचा आयडी.
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 कन्सोलमधील DataFlow टॅबवर जाऊन आमची पाइपलाइन पाहू शकतो. जेव्हा आपण पाइपलाइनवर क्लिक करतो, तेव्हा आपल्याला आकृती 4 सारखे काहीतरी दिसले पाहिजे. डीबगिंगच्या उद्देशाने, तपशीलवार लॉग पाहण्यासाठी लॉग आणि नंतर Stackdriver वर जाणे खूप उपयुक्त ठरू शकते. यामुळे मला अनेक प्रकरणांमध्ये पाइपलाइन समस्यांचे निराकरण करण्यात मदत झाली आहे.

आकृती 4: बीम कन्व्हेयर

BigQuery मध्ये आमच्या डेटामध्ये प्रवेश करा

तर, आमच्या टेबलमध्ये डेटा प्रवाहित करणारी पाइपलाइन आधीपासूनच चालू असावी. याची चाचणी करण्यासाठी, आम्ही BigQuery वर जाऊन डेटा पाहू शकतो. खालील कमांड वापरल्यानंतर तुम्हाला डेटासेटच्या पहिल्या काही पंक्ती दिसल्या पाहिजेत. आता आमच्याकडे BigQuery मध्ये डेटा संग्रहित आहे, आम्ही पुढील विश्लेषण करू शकतो, तसेच सहकार्‍यांसोबत डेटा शेअर करू शकतो आणि व्यावसायिक प्रश्नांची उत्तरे देणे सुरू करू शकतो.

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

आकृती 5: BigQuery

निष्कर्ष

आम्हाला आशा आहे की हे पोस्ट स्ट्रीमिंग डेटा पाइपलाइन तयार करण्यासाठी तसेच डेटा अधिक प्रवेशयोग्य बनवण्याचे मार्ग शोधण्याचे एक उपयुक्त उदाहरण म्हणून काम करेल. या फॉरमॅटमध्ये डेटा संग्रहित केल्याने आम्हाला अनेक फायदे मिळतात. आता आम्ही महत्त्वाच्या प्रश्नांची उत्तरे देऊ शकतो जसे आमचे उत्पादन किती लोक वापरतात? तुमचा वापरकर्ता आधार कालांतराने वाढत आहे का? उत्पादनाच्या कोणत्या पैलूंशी लोक सर्वाधिक संवाद साधतात? आणि जिथे नसावी तिथे त्रुटी आहेत का? हे प्रश्न संस्थेच्या हिताचे असतील. या प्रश्नांच्या उत्तरांमधून उद्भवलेल्या अंतर्दृष्टीच्या आधारावर, आम्ही उत्पादन सुधारू शकतो आणि वापरकर्त्याची प्रतिबद्धता वाढवू शकतो.

बीम या प्रकारच्या व्यायामासाठी खरोखर उपयुक्त आहे आणि इतर अनेक मनोरंजक वापर प्रकरणे देखील आहेत. उदाहरणार्थ, तुम्हाला रिअल टाइममध्ये स्टॉक टिक डेटाचे विश्लेषण करायचे आहे आणि विश्लेषणावर आधारित व्यवहार करायचे आहेत, कदाचित तुमच्याकडे वाहनांमधून येणारा सेन्सर डेटा असेल आणि तुम्हाला ट्रॅफिक पातळीची गणना करायची असेल. उदाहरणार्थ, तुम्ही एक गेमिंग कंपनी देखील असू शकता जी वापरकर्ता डेटा संकलित करते आणि मुख्य मेट्रिक्सचा मागोवा घेण्यासाठी डॅशबोर्ड तयार करण्यासाठी वापरते. ठीक आहे, सज्जनांनो, हा दुसर्‍या पोस्टसाठी विषय आहे, वाचल्याबद्दल धन्यवाद, आणि ज्यांना पूर्ण कोड पहायचा आहे त्यांच्यासाठी खाली माझ्या GitHub ची लिंक आहे.

https://github.com/DFoly/User_log_pipeline

एवढेच. भाग पहिला वाचा.

स्त्रोत: www.habr.com

आम्ही स्ट्रीम डेटा प्रोसेसिंग पाइपलाइन तयार करतो. भाग 2