আমরা একটি স্ট্রিম ডেটা প্রসেসিং পাইপলাইন তৈরি করি। অংশ 2

হাই সব. আমরা নিবন্ধের চূড়ান্ত অংশের অনুবাদ শেয়ার করছি, যা বিশেষভাবে কোর্সের শিক্ষার্থীদের জন্য তৈরি করা হয়েছে। ডেটা ইঞ্জিনিয়ার. আপনি প্রথম অংশ পড়তে পারেন এখানে.

রিয়েল-টাইম পাইপলাইনের জন্য Apache Beam এবং DataFlow

Google ক্লাউড সেট আপ করা হচ্ছে

দ্রষ্টব্য: আমি পাইপলাইন চালাতে এবং কাস্টম লগ ডেটা প্রকাশ করতে Google ক্লাউড শেল ব্যবহার করেছি কারণ পাইথন 3-এ পাইপলাইন চালাতে আমার সমস্যা হচ্ছিল৷ Google ক্লাউড শেল পাইথন 2 ব্যবহার করে, যা Apache Beam-এর সাথে আরও সামঞ্জস্যপূর্ণ৷

পাইপলাইন শুরু করতে, আমাদের সেটিংসে একটু খনন করতে হবে। আপনারা যারা আগে 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. টেবিল লেআউট

ব্যবহারকারীর লগ ডেটা প্রকাশ করা হচ্ছে

পাব/সাব আমাদের পাইপলাইনের একটি গুরুত্বপূর্ণ উপাদান কারণ এটি একাধিক স্বাধীন অ্যাপ্লিকেশনকে একে অপরের সাথে যোগাযোগ করতে দেয়। বিশেষত, এটি একটি মধ্যস্থতাকারী হিসাবে কাজ করে যা আমাদের অ্যাপ্লিকেশনগুলির মধ্যে বার্তা পাঠাতে এবং গ্রহণ করতে দেয়৷ আমাদের প্রথমে একটি বিষয় তৈরি করতে হবে। কনসোলে শুধু পাব/সাব-এ যান এবং টপিক তৈরি করুন ক্লিক করুন।

নীচের কোডটি উপরে সংজ্ঞায়িত লগ ডেটা তৈরি করতে আমাদের স্ক্রিপ্টকে কল করে এবং তারপর লগগুলিকে 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 + সিএটা সম্পূর্ণ করতে

চিত্র 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

এই কমান্ডটি চলাকালীন, আমরা গুগল কনসোলে ডেটাফ্লো ট্যাবে যেতে পারি এবং আমাদের পাইপলাইন দেখতে পারি। যখন আমরা পাইপলাইনে ক্লিক করি, তখন আমাদের চিত্র 4-এর মতো কিছু দেখতে পাওয়া উচিত। ডিবাগিংয়ের উদ্দেশ্যে, বিশদ লগগুলি দেখতে লগ এবং তারপর স্ট্যাকড্রাইভারে যাওয়া খুব সহায়ক হতে পারে। এটি আমাকে বেশ কয়েকটি ক্ষেত্রে পাইপলাইনের সমস্যা সমাধান করতে সাহায্য করেছে।

চিত্র 4: মরীচি পরিবাহক

BigQuery-এ আমাদের ডেটা অ্যাক্সেস করুন

সুতরাং, আমাদের ইতিমধ্যেই একটি পাইপলাইন থাকা উচিত যাতে ডেটা আমাদের টেবিলে প্রবাহিত হয়। এটি পরীক্ষা করার জন্য, আমরা BigQuery-এ যেতে পারি এবং ডেটা দেখতে পারি। নীচের কমান্ডটি ব্যবহার করার পরে আপনি ডেটাসেটের প্রথম কয়েকটি সারি দেখতে পাবেন। এখন যেহেতু আমাদের কাছে BigQuery-এ ডেটা সংরক্ষিত আছে, আমরা আরও বিশ্লেষণ করতে পারি, সেইসাথে সহকর্মীদের সাথে ডেটা শেয়ার করতে পারি এবং ব্যবসায়িক প্রশ্নের উত্তর দেওয়া শুরু করতে পারি।

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

চিত্র 5: BigQuery

উপসংহার

আমরা আশা করি এই পোস্টটি একটি স্ট্রিমিং ডেটা পাইপলাইন তৈরি করার পাশাপাশি ডেটা আরও অ্যাক্সেসযোগ্য করার উপায়গুলি খুঁজে বের করার একটি দরকারী উদাহরণ হিসাবে কাজ করে৷ এই বিন্যাসে ডেটা সংরক্ষণ করা আমাদের অনেক সুবিধা দেয়। এখন আমরা গুরুত্বপূর্ণ প্রশ্নের উত্তর দেওয়া শুরু করতে পারি যেমন কতজন মানুষ আমাদের পণ্য ব্যবহার করছে? আপনার ব্যবহারকারী বেস সময়ের সাথে বাড়ছে? পণ্যের কোন দিকগুলির সাথে লোকেরা সবচেয়ে বেশি যোগাযোগ করে? এবং সেখানে ত্রুটি আছে যেখানে থাকা উচিত নয়? এই প্রশ্নগুলো প্রতিষ্ঠানের স্বার্থের হবে। এই প্রশ্নগুলির উত্তর থেকে উদ্ভূত অন্তর্দৃষ্টিগুলির উপর ভিত্তি করে, আমরা পণ্যটিকে উন্নত করতে এবং ব্যবহারকারীর ব্যস্ততা বাড়াতে পারি।

রশ্মি এই ধরনের ব্যায়ামের জন্য সত্যিই উপযোগী এবং অন্যান্য আকর্ষণীয় ব্যবহারের ক্ষেত্রেও রয়েছে। উদাহরণস্বরূপ, আপনি রিয়েল টাইমে স্টক টিক ডেটা বিশ্লেষণ করতে এবং বিশ্লেষণের উপর ভিত্তি করে ব্যবসা করতে চাইতে পারেন, সম্ভবত আপনার কাছে যানবাহন থেকে সেন্সর ডেটা আসছে এবং আপনি ট্র্যাফিক স্তরের গণনা করতে চান। উদাহরণস্বরূপ, আপনি একটি গেমিং কোম্পানি হতে পারেন যা ব্যবহারকারীর ডেটা সংগ্রহ করে এবং মূল মেট্রিক্স ট্র্যাক করতে ড্যাশবোর্ড তৈরি করতে এটি ব্যবহার করে। ঠিক আছে, ভদ্রলোক, এটি অন্য পোস্টের জন্য একটি বিষয়, পড়ার জন্য ধন্যবাদ, এবং যারা সম্পূর্ণ কোড দেখতে চান তাদের জন্য, নীচে আমার গিটহাবের লিঙ্ক রয়েছে।

https://github.com/DFoly/User_log_pipeline

এখানেই শেষ. প্রথম অংশ পড়ুন.

উত্স: www.habr.com