हामी स्ट्रिम डाटा प्रोसेसिङ पाइपलाइन सिर्जना गर्छौं। भाग 2

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

Apache Beam र DataFlow वास्तविक-समय पाइपलाइनहरूको लागि

Google क्लाउड सेटअप गर्दै

नोट: मैले पाइपलाइन चलाउन र अनुकूलन लग डेटा प्रकाशित गर्न Google क्लाउड शेल प्रयोग गरें किनभने मलाई पाइथन 3 मा पाइपलाइन चलाउन समस्या भइरहेको थियो। Google क्लाउड शेलले पाइथन 2 प्रयोग गर्दछ, जुन Apache Beam सँग अधिक सुसंगत छ।

पाइपलाइन सुरु गर्न, हामीले सेटिङहरूमा थोरै खन्नुपर्छ। तपाईंहरू मध्ये जसले पहिले GCP प्रयोग गर्नुभएको छैन, तपाईंले यसमा उल्लिखित निम्न 6 चरणहरू पालना गर्नुपर्नेछ। पृष्ठ.

यस पछि, हामीले हाम्रा स्क्रिप्टहरू गुगल क्लाउड भण्डारणमा अपलोड गर्नुपर्नेछ र तिनीहरूलाई हाम्रो गुगल क्लाउड शेलमा प्रतिलिपि गर्न आवश्यक छ। क्लाउड भण्डारणमा अपलोड गर्नु एकदम मामूली छ (एक विवरण फेला पार्न सकिन्छ यहाँ)। हाम्रा फाइलहरू प्रतिलिपि गर्न, हामी तलको चित्र २ मा बाँयामा रहेको पहिलो आइकनमा क्लिक गरेर उपकरणपट्टीबाट गुगल क्लाउड शेल खोल्न सक्छौं।

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 क्लाउड कन्सोल प्रयोग गर्नु हो। तपाईं तलका चरणहरू पालना गर्न सक्नुहुन्छ लिङ्कस्कीमाको साथ तालिका सिर्जना गर्न। हाम्रो टेबल हुनेछ १२ स्तम्भहरू, प्रत्येक प्रयोगकर्ता लग को अवयवहरु संगत। सुविधाको लागि, हामी टाइमलोकल भेरिएबल बाहेक सबै स्तम्भहरूलाई स्ट्रिङको रूपमा परिभाषित गर्नेछौं, र हामीले पहिले उत्पन्न गरेका चरहरू अनुसार तिनीहरूलाई नाम दिनेछौं। हाम्रो तालिकाको लेआउट चित्र 3 मा देखिनु पर्छ।

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

प्रयोगकर्ता लग डेटा प्रकाशित गर्दै

Pub/Sub हाम्रो पाइपलाइनको एक महत्वपूर्ण भाग हो किनभने यसले धेरै स्वतन्त्र अनुप्रयोगहरूलाई एकअर्कासँग सञ्चार गर्न अनुमति दिन्छ। विशेष गरी, यसले एक मध्यस्थको रूपमा काम गर्दछ जसले हामीलाई अनुप्रयोगहरू बीच सन्देशहरू पठाउन र प्राप्त गर्न अनुमति दिन्छ। हामीले गर्नु पर्ने पहिलो कुरा भनेको विषय सिर्जना गर्नु हो। केवल कन्सोलमा पब/सबमा जानुहोस् र विषय सिर्जना गर्नुहोस् क्लिक गर्नुहोस्।

तलको कोडले माथि परिभाषित लग डेटा उत्पन्न गर्न हाम्रो स्क्रिप्टलाई कल गर्छ र त्यसपछि लगहरूलाई Pub/Sub मा जडान गरी पठाउँछ। हामीले गर्नु पर्ने कुरा भनेको वस्तु सिर्जना गर्नु हो प्रकाशक ग्राहक, विधि प्रयोग गरी विषयको मार्ग निर्दिष्ट गर्नुहोस् topic_path र समारोहमा कल गर्नुहोस् publish с topic_path र डाटा। कृपया ध्यान दिनुहोस् कि हामी आयात गर्छौं generate_log_line हाम्रो लिपिबाट stream_logs, त्यसैले यी फाइलहरू एउटै फोल्डरमा छन् भन्ने निश्चित गर्नुहोस्, अन्यथा तपाईंले आयात त्रुटि पाउनुहुनेछ। त्यसपछि हामी यसलाई हाम्रो गुगल कन्सोल मार्फत चलाउन सक्छौं:

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यसलाई पूरा गर्न।

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

हाम्रो पाइपलाइन कोड लेख्दै

अब जब हामीसँग सबै तयार छ, हामी रमाइलो भाग सुरु गर्न सक्छौं - बीम र पाइथन प्रयोग गरेर हाम्रो पाइपलाइन कोडिङ। बीम पाइपलाइन सिर्जना गर्न, हामीले पाइपलाइन वस्तु (p) सिर्जना गर्न आवश्यक छ। एकपटक हामीले पाइपलाइन वस्तु सिर्जना गरिसकेपछि, हामी अपरेटर प्रयोग गरेर एकपछि अर्को धेरै प्रकार्यहरू लागू गर्न सक्छौं pipe (|)। सामान्यतया, कार्यप्रवाह तलको छवि जस्तो देखिन्छ।

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

हाम्रो कोडमा, हामी दुई अनुकूलन प्रकार्यहरू सिर्जना गर्नेछौं। समारोह regex_clean, जसले डेटा स्क्यान गर्दछ र प्रकार्य प्रयोग गरेर PATTERNS सूचीमा आधारित सम्बन्धित पङ्क्ति पुन: प्राप्त गर्दछ। re.search। प्रकार्यले अल्पविराम छुट्याएको स्ट्रिङ फर्काउँछ। यदि तपाईं नियमित अभिव्यक्ति विशेषज्ञ हुनुहुन्न भने, म यसलाई जाँच गर्न सिफारिस गर्छु ट्यूटोरियल र कोड जाँच गर्न नोटप्याडमा अभ्यास गर्नुहोस्। यस पछि हामी कस्टम ParDo प्रकार्य परिभाषित गर्छौं विभाजन, जुन समानान्तर प्रशोधनको लागि बीम रूपान्तरणको भिन्नता हो। पाइथनमा, यो एक विशेष तरिकामा गरिन्छ - हामीले DoFn बीम वर्गबाट ​​इनहेरिट गर्ने कक्षा सिर्जना गर्नुपर्छ। स्प्लिट प्रकार्यले अघिल्लो प्रकार्यबाट पार्स गरिएको पङ्क्ति लिन्छ र हाम्रो BigQuery तालिकामा स्तम्भ नामहरूसँग सम्बन्धित कुञ्जीहरू सहित शब्दकोशहरूको सूची फर्काउँछ। यस प्रकार्यको बारेमा नोट गर्न केहि छ: मैले आयात गर्नुपर्‍यो datetime यसलाई काम गर्न को लागी एक प्रकार्य भित्र। मैले फाइलको सुरुमा आयात त्रुटि प्राप्त गरिरहेको थिएँ, जुन अनौठो थियो। यो सूची त्यसपछि समारोहमा पठाइन्छ BigQuery लेख्नुहोस्, जसले हाम्रो डाटालाई टेबलमा थप्छ। ब्याच DataFlow Job र Streaming DataFlow Job को लागि कोड तल दिइएको छ। ब्याच र स्ट्रिमिङ कोड बीचको भिन्नता भनेको ब्याचमा हामीले 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 परियोजनाको ID।
• 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;

चित्र ५: BigQuery

निष्कर्षमा

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

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

https://github.com/DFoly/User_log_pipeline

यति नै। भाग एक पढ्नुहोस्.

स्रोत: www.habr.com