ഞങ്ങൾ ഒരു സ്ട്രീം ഡാറ്റ പ്രോസസ്സിംഗ് പൈപ്പ്ലൈൻ സൃഷ്ടിക്കുന്നു. ഭാഗം 2

എല്ലാവർക്കും ഹായ്. കോഴ്‌സിലെ വിദ്യാർത്ഥികൾക്കായി പ്രത്യേകം തയ്യാറാക്കിയ ലേഖനത്തിന്റെ അവസാന ഭാഗത്തിന്റെ വിവർത്തനം ഞങ്ങൾ പങ്കിടുന്നു. "ഡാറ്റ എഞ്ചിനീയർ". ആദ്യഭാഗം വായിക്കാം ഇവിടെ.

തത്സമയ പൈപ്പ്ലൈനുകൾക്കായി അപ്പാച്ചെ ബീമും ഡാറ്റാഫ്ലോയും

Google ക്ലൗഡ് സജ്ജീകരിക്കുന്നു

ശ്രദ്ധിക്കുക: പൈത്തൺ 3-ൽ പൈപ്പ്‌ലൈൻ പ്രവർത്തിപ്പിക്കുന്നതിൽ എനിക്ക് പ്രശ്‌നമുണ്ടായതിനാൽ പൈപ്പ്‌ലൈൻ പ്രവർത്തിപ്പിക്കുന്നതിനും ഇഷ്‌ടാനുസൃത ലോഗ് ഡാറ്റ പ്രസിദ്ധീകരിക്കുന്നതിനും ഞാൻ Google ക്ലൗഡ് ഷെൽ ഉപയോഗിച്ചു.

പൈപ്പ്ലൈൻ ആരംഭിക്കുന്നതിന്, ഞങ്ങൾ ക്രമീകരണങ്ങളിൽ അൽപ്പം കുഴിക്കേണ്ടതുണ്ട്. നിങ്ങളിൽ മുമ്പ് 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, അതിനാൽ ഈ ഫയലുകൾ ഒരേ ഫോൾഡറിലാണെന്ന് ഉറപ്പാക്കുക, അല്ലാത്തപക്ഷം നിങ്ങൾക്ക് ഒരു ഇറക്കുമതി പിശക് ലഭിക്കും. പിന്നീട് ഇത് ഉപയോഗിച്ച് ഞങ്ങളുടെ ഗൂഗിൾ കൺസോളിലൂടെ ഇത് പ്രവർത്തിപ്പിക്കാം:

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

ഞങ്ങളുടെ പൈപ്പ്ലൈൻ കോഡ് എഴുതുന്നു

ഇപ്പോൾ ഞങ്ങൾ എല്ലാം തയ്യാറാക്കിയിട്ടുണ്ട്, നമുക്ക് രസകരമായ ഭാഗം ആരംഭിക്കാം - ബീം, പൈത്തൺ എന്നിവ ഉപയോഗിച്ച് ഞങ്ങളുടെ പൈപ്പ്ലൈൻ കോഡിംഗ്. ഒരു ബീം പൈപ്പ്ലൈൻ സൃഷ്ടിക്കുന്നതിന്, ഞങ്ങൾ ഒരു പൈപ്പ്ലൈൻ ഒബ്ജക്റ്റ് (p) സൃഷ്ടിക്കേണ്ടതുണ്ട്. ഒരു പൈപ്പ് ലൈൻ ഒബ്‌ജക്‌റ്റ് സൃഷ്‌ടിച്ചുകഴിഞ്ഞാൽ, ഓപ്പറേറ്റർ ഉപയോഗിച്ച് നമുക്ക് ഒന്നിലധികം ഫംഗ്‌ഷനുകൾ പ്രയോഗിക്കാൻ കഴിയും pipe (|). പൊതുവേ, വർക്ക്ഫ്ലോ ചുവടെയുള്ള ചിത്രം പോലെ കാണപ്പെടുന്നു.

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

ഞങ്ങളുടെ കോഡിൽ, ഞങ്ങൾ രണ്ട് ഇഷ്‌ടാനുസൃത പ്രവർത്തനങ്ങൾ സൃഷ്ടിക്കും. ഫംഗ്ഷൻ regex_clean, അത് ഡാറ്റ സ്കാൻ ചെയ്യുകയും ഫംഗ്ഷൻ ഉപയോഗിച്ച് പാറ്റേൺസ് ലിസ്റ്റിനെ അടിസ്ഥാനമാക്കി അനുബന്ധ വരി വീണ്ടെടുക്കുകയും ചെയ്യുന്നു re.search. ഫംഗ്ഷൻ കോമയാൽ വേർതിരിച്ച ഒരു സ്ട്രിംഗ് നൽകുന്നു. നിങ്ങൾ ഒരു സാധാരണ എക്സ്പ്രഷൻ വിദഗ്ദ്ധനല്ലെങ്കിൽ, ഇത് പരിശോധിക്കാൻ ഞാൻ ശുപാർശ ചെയ്യുന്നു ട്യൂട്ടോറിയൽ കോഡ് പരിശോധിക്കാൻ ഒരു നോട്ട്പാഡിൽ പരിശീലിക്കുക. ഇതിനുശേഷം ഞങ്ങൾ ഒരു ഇഷ്‌ടാനുസൃത ParDo ഫംഗ്‌ഷൻ നിർവചിക്കുന്നു രണ്ടായി പിരിയുക, സമാന്തര പ്രോസസ്സിംഗിനുള്ള ബീം പരിവർത്തനത്തിന്റെ ഒരു വ്യതിയാനമാണിത്. പൈത്തണിൽ, ഇത് ഒരു പ്രത്യേക രീതിയിലാണ് ചെയ്യുന്നത് - DoFn ബീം ക്ലാസിൽ നിന്ന് പാരമ്പര്യമായി ലഭിക്കുന്ന ഒരു ക്ലാസ് ഞങ്ങൾ സൃഷ്ടിക്കണം. സ്പ്ലിറ്റ് ഫംഗ്‌ഷൻ മുമ്പത്തെ ഫംഗ്‌ഷനിൽ നിന്ന് പാഴ്‌സ് ചെയ്‌ത വരി എടുക്കുകയും ഞങ്ങളുടെ BigQuery പട്ടികയിലെ കോളം പേരുകൾക്ക് അനുയോജ്യമായ കീകളുള്ള നിഘണ്ടുക്കളുടെ ഒരു ലിസ്റ്റ് നൽകുകയും ചെയ്യുന്നു. ഈ ഫംഗ്‌ഷനിൽ ശ്രദ്ധിക്കേണ്ട ചിലതുണ്ട്: എനിക്ക് ഇറക്കുമതി ചെയ്യേണ്ടിവന്നു datetime ഒരു ഫംഗ്‌ഷന്റെ ഉള്ളിൽ അത് പ്രവർത്തിക്കുന്നു. ഫയലിന്റെ തുടക്കത്തിൽ എനിക്ക് ഒരു ഇറക്കുമതി പിശക് ലഭിക്കുന്നു, അത് വിചിത്രമായിരുന്നു. ഈ ലിസ്റ്റ് പിന്നീട് ഫംഗ്ഷനിലേക്ക് കൈമാറും WriteToBigQuery, ഇത് ഞങ്ങളുടെ ഡാറ്റയെ പട്ടികയിലേക്ക് ചേർക്കുന്നു. Batch 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()

കൺവെയർ ആരംഭിക്കുന്നു

പൈപ്പ് ലൈൻ നമുക്ക് പല തരത്തിൽ പ്രവർത്തിപ്പിക്കാം. ഞങ്ങൾക്ക് വേണമെങ്കിൽ, ജിസിപിയിലേക്ക് വിദൂരമായി ലോഗിൻ ചെയ്യുമ്പോൾ ടെർമിനലിൽ നിന്ന് പ്രാദേശികമായി പ്രവർത്തിപ്പിക്കാം.

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-ന് സമാനമായ ഒന്ന് നമ്മൾ കാണും. ഡീബഗ്ഗിംഗ് ആവശ്യങ്ങൾക്കായി, വിശദമായ ലോഗുകൾ കാണുന്നതിന് ലോഗുകളിലേക്കും തുടർന്ന് 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