અમે સ્ટ્રીમ ડેટા પ્રોસેસિંગ પાઇપલાઇન બનાવીએ છીએ. ભાગ 2

કેમ છો બધા. અમે લેખના અંતિમ ભાગનો અનુવાદ શેર કરી રહ્યા છીએ, જે ખાસ કરીને કોર્સના વિદ્યાર્થીઓ માટે તૈયાર કરવામાં આવ્યો છે. ડેટા એન્જિનિયર. તમે પ્રથમ ભાગ વાંચી શકો છો અહીં.

રીઅલ-ટાઇમ પાઇપલાઇન્સ માટે અપાચે બીમ અને ડેટાફ્લો

Google Cloud સેટ કરી રહ્યાં છીએ

નોંધ: મેં પાઇપલાઇન ચલાવવા અને કસ્ટમ લોગ ડેટા પ્રકાશિત કરવા માટે Google ક્લાઉડ શેલનો ઉપયોગ કર્યો છે કારણ કે મને પાયથોન 3 માં પાઇપલાઇન ચલાવવામાં મુશ્કેલી આવી રહી હતી. Google ક્લાઉડ શેલ પાયથોન 2 નો ઉપયોગ કરે છે, જે અપાચે બીમ સાથે વધુ સુસંગત છે.

પાઇપલાઇન શરૂ કરવા માટે, આપણે સેટિંગ્સમાં થોડું ખોદવું પડશે. તમારામાંથી જેમણે અગાઉ GCP નો ઉપયોગ કર્યો નથી, તમારે આમાં દર્શાવેલ નીચેના 6 પગલાંને અનુસરવાની જરૂર પડશે પૃષ્ઠ.

આ પછી, અમારે અમારી સ્ક્રિપ્ટ્સને Google Cloud Storage પર અપલોડ કરવાની અને તેને અમારા Google Cloud Shel પર કૉપિ કરવાની જરૂર પડશે. ક્લાઉડ સ્ટોરેજ પર અપલોડ કરવું એકદમ તુચ્છ છે (વર્ણન મળી શકે છે અહીં). અમારી ફાઇલોની નકલ કરવા માટે, અમે નીચે આકૃતિ 2 માં ડાબી બાજુના પ્રથમ આઇકોન પર ક્લિક કરીને ટૂલબારમાંથી Google Cloud Shel ખોલી શકીએ છીએ.

ચિત્ર 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 Cloud કન્સોલનો ઉપયોગ કરવો. તમે નીચેના પગલાંને અનુસરી શકો છો કડીસ્કીમા સાથે ટેબલ બનાવવા માટે. અમારા ટેબલ હશે 7 કૉલમ, દરેક વપરાશકર્તા લોગના ઘટકોને અનુરૂપ. સગવડ માટે, અમે ટાઈમલોકલ વેરીએબલ સિવાયના તમામ કૉલમ્સને સ્ટ્રિંગ તરીકે વ્યાખ્યાયિત કરીશું અને અમે અગાઉ જનરેટ કરેલા વેરિયેબલ્સ અનુસાર તેમને નામ આપીશું. અમારા કોષ્ટકનું લેઆઉટ આકૃતિ 3 જેવું હોવું જોઈએ.

આકૃતિ 3. ટેબલ લેઆઉટ

વપરાશકર્તા લોગ ડેટા પ્રકાશિત કરી રહ્યું છે

પબ/સબ એ અમારી પાઇપલાઇનનો એક મહત્વપૂર્ણ ઘટક છે કારણ કે તે બહુવિધ સ્વતંત્ર એપ્લિકેશનોને એકબીજા સાથે વાતચીત કરવાની મંજૂરી આપે છે. ખાસ કરીને, તે મધ્યસ્થી તરીકે કામ કરે છે જે અમને એપ્લિકેશનો વચ્ચે સંદેશા મોકલવા અને પ્રાપ્ત કરવાની મંજૂરી આપે છે. પ્રથમ વસ્તુ જે આપણે કરવાની જરૂર છે તે એક વિષય બનાવવાની છે. કન્સોલમાં ફક્ત Pub/Sub પર જાઓ અને CREATE TOPIC પર ક્લિક કરો.

નીચેનો કોડ ઉપર વ્યાખ્યાયિત લોગ ડેટા જનરેટ કરવા માટે અમારી સ્ક્રિપ્ટને કૉલ કરે છે અને પછી લોગને 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)

જલદી ફાઇલ ચાલે છે, અમે નીચેની આકૃતિમાં બતાવ્યા પ્રમાણે, કન્સોલ પર લોગ ડેટાનું આઉટપુટ જોઈ શકીશું. જ્યાં સુધી અમે ઉપયોગ નહીં કરીએ ત્યાં સુધી આ સ્ક્રિપ્ટ કામ કરશે સીટીઆરએલ + સીતેને પૂર્ણ કરવા માટે.

આકૃતિ 4. આઉટપુટ 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 તેને કાર્ય કરવા માટે કાર્યની અંદર. મને ફાઇલની શરૂઆતમાં એક આયાત ભૂલ મળી રહી હતી, જે વિચિત્ર હતી. આ યાદી પછી ફંક્શનમાં મોકલવામાં આવે છે બિગક્વેરી લખો, જે ફક્ત ટેબલમાં આપણો ડેટા ઉમેરે છે. બેચ ડેટાફ્લો જોબ અને સ્ટ્રીમિંગ ડેટાફ્લો જોબ માટેનો કોડ નીચે આપેલ છે. બેચ અને સ્ટ્રીમિંગ કોડ વચ્ચેનો તફાવત એટલો જ છે કે બેચમાં આપણે 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

જો કે, અમે ડેટાફ્લોનો ઉપયોગ કરીને તેને ચલાવવા જઈ રહ્યા છીએ. અમે નીચેના જરૂરી પરિમાણો સેટ કરીને નીચેના આદેશનો ઉપયોગ કરીને આ કરી શકીએ છીએ.

• project — તમારા GCP પ્રોજેક્ટનું ID.
• runner એક પાઇપલાઇન રનર છે જે તમારા પ્રોગ્રામનું વિશ્લેષણ કરશે અને તમારી પાઇપલાઇનનું નિર્માણ કરશે. ક્લાઉડમાં ચલાવવા માટે, તમારે ડેટાફ્લોરનરનો ઉલ્લેખ કરવો આવશ્યક છે.
• 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 જેવું કંઈક જોવું જોઈએ. ડીબગીંગ હેતુઓ માટે, વિગતવાર લોગ જોવા માટે લોગ્સ અને પછી સ્ટેકડ્રાઈવર પર જવાનું ખૂબ જ મદદરૂપ થઈ શકે છે. આનાથી મને અસંખ્ય કેસોમાં પાઇપલાઇન સમસ્યાઓ ઉકેલવામાં મદદ મળી છે.

આકૃતિ 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