យើងបង្កើតបំពង់ដំណើរការទិន្នន័យស្ទ្រីម។ ផ្នែកទី 2

សួស្តី​អ្នក​ទាំងអស់គ្នា។ យើងកំពុងចែករំលែកការបកប្រែនៃផ្នែកចុងក្រោយនៃអត្ថបទ ដែលត្រូវបានរៀបចំជាពិសេសសម្រាប់សិស្សនៃវគ្គសិក្សា។ "វិស្វករទិន្នន័យ". អ្នកអាចអានផ្នែកដំបូង នៅទីនេះ.

Apache Beam និង DataFlow សម្រាប់បំពង់បង្ហូរប្រេងពេលវេលាពិត

ការដំឡើង Google Cloud

ចំណាំ៖ ខ្ញុំបានប្រើ Google Cloud Shell ដើម្បីដំណើរការបំពង់បង្ហូរប្រេង និងបោះផ្សាយទិន្នន័យកំណត់ហេតុផ្ទាល់ខ្លួន ដោយសារខ្ញុំមានបញ្ហាក្នុងការដំណើរការបំពង់បង្ហូរប្រេងនៅក្នុង Python 3។ Google Cloud Shell ប្រើ Python 2 ដែលសមស្របជាងជាមួយ Apache Beam ។

ដើម្បីចាប់ផ្តើមបំពង់បង្ហូរប្រេងយើងត្រូវជីកបន្តិចទៅក្នុងការកំណត់។ សម្រាប់​អ្នក​ដែល​មិន​ធ្លាប់​ប្រើ GCP ពីមុន អ្នក​នឹង​ត្រូវ​ធ្វើ​តាម​ជំហាន 6 ខាងក្រោម​ដែល​បាន​រៀបរាប់​នៅក្នុង​នេះ។ ទំព័រ.

បន្ទាប់ពីនេះ យើងនឹងត្រូវបង្ហោះស្គ្រីបរបស់យើងទៅ Google Cloud Storage ហើយចម្លងវាទៅ Google Cloud Shel របស់យើង។ ការ​ផ្ទុក​ឡើង​ទៅ​កន្លែង​ផ្ទុក​ពពក​គឺ​ជា​រឿង​តូចតាច​ណាស់ (អាច​រក​ឃើញ​ការ​ពិពណ៌នា នៅទីនេះ) ដើម្បីចម្លងឯកសាររបស់យើង យើងអាចបើក Google Cloud Shel ពីរបារឧបករណ៍ដោយចុចលើរូបតំណាងទីមួយនៅខាងឆ្វេងក្នុងរូបភាពទី 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 Cloud ដោយបង្កើតសំណុំទិន្នន័យជាមុនសិន។ អ្នកអាចអនុវត្តតាមជំហានខាងក្រោម តំណភ្ជាប់ដើម្បីបង្កើតតារាងជាមួយគ្រោងការណ៍។ តុរបស់យើងនឹងមាន 7 ជួរដែលត្រូវគ្នាទៅនឹងសមាសធាតុនៃកំណត់ហេតុអ្នកប្រើប្រាស់នីមួយៗ។ ដើម្បីភាពងាយស្រួល យើងនឹងកំណត់ជួរឈរទាំងអស់ជាខ្សែអក្សរ លើកលែងតែអថេរតាមកាលកំណត់ ហើយដាក់ឈ្មោះពួកវាតាមអថេរដែលយើងបានបង្កើតមុននេះ។ ប្លង់នៃតារាងរបស់យើងគួរតែមើលទៅដូចក្នុងរូបភាពទី 3 ។

រូបភាពទី 3. ប្លង់តារាង

ការបោះពុម្ពទិន្នន័យកំណត់ហេតុអ្នកប្រើប្រាស់

Pub/Sub គឺជាធាតុផ្សំដ៏សំខាន់នៃបំពង់បង្ហូរប្រេងរបស់យើង ព្រោះវាអនុញ្ញាតឱ្យកម្មវិធីឯករាជ្យជាច្រើនទាក់ទងគ្នាទៅវិញទៅមក។ ជាពិសេស វាធ្វើការជាអន្តរការីដែលអនុញ្ញាតឱ្យយើងផ្ញើ និងទទួលសាររវាងកម្មវិធី។ រឿងដំបូងដែលយើងត្រូវធ្វើគឺបង្កើតប្រធានបទ។ គ្រាន់តែចូលទៅកាន់ Pub/Sub នៅក្នុងកុងសូល ហើយចុច បង្កើតប្រធានបទ។

កូដខាងក្រោមហៅស្គ្រីបរបស់យើងដើម្បីបង្កើតទិន្នន័យកំណត់ហេតុដែលបានកំណត់ខាងលើ ហើយបន្ទាប់មកភ្ជាប់ និងផ្ញើកំណត់ហេតុទៅ Pub/Sub ។ រឿងតែមួយគត់ដែលយើងត្រូវធ្វើគឺបង្កើតវត្ថុមួយ។ អ្នកបោះពុម្ពផ្សាយបញ្ជាក់ផ្លូវទៅកាន់ប្រធានបទដោយប្រើវិធីសាស្ត្រ 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 + Cដើម្បីបំពេញវា។

រូបភាពទី 4. ទិន្នផល publish_logs.py

ការសរសេរកូដបំពង់របស់យើង។

ឥឡូវនេះ យើងបានរៀបចំអ្វីៗគ្រប់យ៉ាងហើយ យើងអាចចាប់ផ្តើមផ្នែកដ៏រីករាយ ពោលគឺការសរសេរកូដបំពង់របស់យើងដោយប្រើ Beam និង Python ។ ដើម្បីបង្កើត Beam pipeline យើងត្រូវបង្កើត pipeline object (p)។ នៅពេលដែលយើងបានបង្កើតវត្ថុបំពង់មួយ យើងអាចអនុវត្តមុខងារច្រើនមួយបន្ទាប់ពីមួយផ្សេងទៀតដោយប្រើប្រតិបត្តិករ pipe (|). ជាទូទៅលំហូរការងារមើលទៅដូចរូបភាពខាងក្រោម។

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

នៅក្នុងកូដរបស់យើង យើងនឹងបង្កើតមុខងារផ្ទាល់ខ្លួនពីរ។ មុខងារ regex_cleanដែលស្កែនទិន្នន័យ និងទាញយកជួរដេកដែលត្រូវគ្នាដោយផ្អែកលើបញ្ជី PATTERNS ដោយប្រើមុខងារ re.search. អនុគមន៍​ត្រឡប់​ខ្សែអក្សរ​ដែលបំបែកដោយ​សញ្ញាក្បៀស។ ប្រសិនបើអ្នកមិនមែនជាអ្នកជំនាញខាងការបញ្ចេញមតិធម្មតាទេ ខ្ញុំសូមណែនាំឱ្យពិនិត្យមើលរឿងនេះ ការបង្រៀន ហើយអនុវត្តនៅក្នុង notepad ដើម្បីពិនិត្យមើលកូដ។ បន្ទាប់ពីនេះយើងកំណត់មុខងារ ParDo ផ្ទាល់ខ្លួនហៅថា ពុះដែលជាបំរែបំរួលនៃការផ្លាស់ប្តូរ Beam សម្រាប់ដំណើរការប៉ារ៉ាឡែល។ នៅក្នុង Python នេះត្រូវបានធ្វើតាមរបៀបពិសេសមួយ - យើងត្រូវបង្កើត class ដែលទទួលមរតកពី DoFn Beam class ។ មុខងារបំបែកយកជួរដែលបានញែកចេញពីមុខងារមុន ហើយត្រឡប់បញ្ជីវចនានុក្រមដែលមានគ្រាប់ចុចដែលត្រូវគ្នានឹងឈ្មោះជួរឈរនៅក្នុងតារាង BigQuery របស់យើង។ មានអ្វីដែលត្រូវកត់សម្គាល់អំពីមុខងារនេះ៖ ខ្ញុំត្រូវនាំចូល datetime នៅខាងក្នុងមុខងារដើម្បីធ្វើឱ្យវាដំណើរការ។ ខ្ញុំ​ទទួល​បាន​កំហុស​ការ​នាំ​ចូល​នៅ​ដើម​ឯកសារ​ដែល​ចម្លែក។ បន្ទាប់មកបញ្ជីនេះត្រូវបានបញ្ជូនទៅមុខងារ WriteToBigQueryដែលគ្រាន់តែបន្ថែមទិន្នន័យរបស់យើងទៅក្នុងតារាង។ កូដសម្រាប់ Batch DataFlow Job និង Streaming DataFlow Job ត្រូវបានផ្តល់ឱ្យខាងក្រោម។ ភាពខុសគ្នាតែមួយគត់រវាង batch និង streaming code គឺថានៅក្នុង batch យើងអាន CSV ពី src_pathដោយប្រើមុខងារ ReadFromText ពី Beam ។

Batch DataFlow Job (ដំណើរការជាបាច់)

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()

ការងារស្ទ្រីម DataFlow (ដំណើរការស្ទ្រីម)

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 - ផ្លូវទៅកាន់ Cloud Dataflow cloud storage សម្រាប់បង្កើតលិបិក្រមកញ្ចប់កូដដែលត្រូវការដោយ processors ដែលដំណើរការការងារ។
• temp_location — ផ្លូវទៅកាន់ Cloud Dataflow cloud storage សម្រាប់រក្សាទុកឯកសារការងារបណ្ដោះអាសន្នដែលបានបង្កើតខណៈពេលដែលបំពង់កំពុងដំណើរការ។
• streaming

python main_pipeline_stream.py 
--runner DataFlow 
--project $PROJECT 
--temp_location $BUCKET/tmp 
--staging_location $BUCKET/staging
--streaming

ខណៈពេលដែលពាក្យបញ្ជានេះកំពុងដំណើរការ យើងអាចចូលទៅកាន់ផ្ទាំង DataFlow នៅក្នុងកុងសូល Google ហើយមើលបំពង់របស់យើង។ នៅពេលដែលយើងចុចលើបំពង់បង្ហូរប្រេង យើងគួរតែឃើញអ្វីដែលស្រដៀងនឹងរូបភាពទី 4។ សម្រាប់គោលបំណងបំបាត់កំហុស វាអាចមានប្រយោជន៍ខ្លាំងណាស់ក្នុងការចូលទៅកាន់ Logs ហើយបន្ទាប់មកទៅកាន់ Stackdriver ដើម្បីមើលកំណត់ហេតុលម្អិត។ នេះបានជួយខ្ញុំដោះស្រាយបញ្ហាបំពង់បង្ហូរប្រេងនៅក្នុងករណីមួយចំនួន។

រូបភាពទី 4: ឧបករណ៍បញ្ជូនធ្នឹម

ចូលប្រើទិន្នន័យរបស់យើងនៅក្នុង BigQuery

ដូច្នេះ យើងគួរតែមានបំពង់បង្ហូរប្រេងដែលដំណើរការជាមួយទិន្នន័យដែលហូរចូលទៅក្នុងតារាងរបស់យើង។ ដើម្បីសាកល្បងនេះ យើងអាចចូលទៅកាន់ BigQuery ហើយមើលទិន្នន័យ។ បន្ទាប់ពីប្រើពាក្យបញ្ជាខាងក្រោម អ្នកគួរតែឃើញជួរពីរបីដំបូងនៃសំណុំទិន្នន័យ។ ឥឡូវនេះយើងមានទិន្នន័យដែលបានរក្សាទុកនៅក្នុង BigQuery យើងអាចធ្វើការវិភាគបន្ថែម ក៏ដូចជាចែករំលែកទិន្នន័យជាមួយសហសេវិក ហើយចាប់ផ្តើមឆ្លើយសំណួរអាជីវកម្ម។

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

រូបភាពទី 5៖ BigQuery

សេចក្តីសន្និដ្ឋាន

យើងសង្ឃឹមថាការបង្ហោះនេះដើរតួជាឧទាហរណ៍ដ៏មានប្រយោជន៍នៃការបង្កើតបំពង់បញ្ជូនទិន្នន័យ ក៏ដូចជាការស្វែងរកវិធីធ្វើឱ្យទិន្នន័យអាចចូលប្រើប្រាស់បានកាន់តែច្រើន។ ការរក្សាទុកទិន្នន័យក្នុងទម្រង់នេះផ្តល់ឱ្យយើងនូវអត្ថប្រយោជន៍ជាច្រើន។ ឥឡូវនេះយើងអាចចាប់ផ្តើមឆ្លើយសំណួរសំខាន់ៗដូចជាតើមានមនុស្សប៉ុន្មាននាក់ដែលប្រើផលិតផលរបស់យើង? តើមូលដ្ឋានអ្នកប្រើប្រាស់របស់អ្នកកើនឡើងតាមពេលវេលាដែរឬទេ? តើទិដ្ឋភាពអ្វីខ្លះនៃផលិតផលដែលមនុស្សទាក់ទងគ្នាច្រើនបំផុត? ហើយមានកំហុសត្រង់ណាដែលមិនគួរមាន? ទាំងនេះគឺជាសំណួរដែលនឹងចាប់អារម្មណ៍ចំពោះអង្គការ។ ដោយផ្អែកលើការយល់ដឹងដែលផុសចេញពីចម្លើយចំពោះសំណួរទាំងនេះ យើងអាចកែលម្អផលិតផល និងបង្កើនការចូលរួមរបស់អ្នកប្រើប្រាស់។

Beam ពិតជាមានប្រយោជន៍សម្រាប់លំហាត់ប្រភេទនេះ ហើយមានករណីប្រើប្រាស់គួរឱ្យចាប់អារម្មណ៍មួយចំនួនទៀតផងដែរ។ ឧទាហរណ៍ អ្នកប្រហែលជាចង់វិភាគទិន្នន័យសញ្ញាធីកក្នុងពេលវេលាជាក់ស្តែង និងធ្វើពាណិជ្ជកម្មដោយផ្អែកលើការវិភាគ ប្រហែលជាអ្នកមានទិន្នន័យឧបករណ៍ចាប់សញ្ញាមកពីយានយន្ត ហើយចង់គណនាការគណនាកម្រិតចរាចរណ៍។ ឧទាហរណ៍ អ្នកក៏អាចជាក្រុមហ៊ុនហ្គេមដែលប្រមូលទិន្នន័យអ្នកប្រើប្រាស់ ហើយប្រើវាដើម្បីបង្កើតផ្ទាំងគ្រប់គ្រងដើម្បីតាមដានម៉ែត្រសំខាន់ៗ។ អូខេ សុភាពនារី នេះជាប្រធានបទសម្រាប់ប្រកាសមួយទៀត អរគុណសម្រាប់ការអាន ហើយសម្រាប់អ្នកចង់មើលកូដពេញ ខាងក្រោមនេះជាតំណភ្ជាប់ទៅកាន់ GitHub របស់ខ្ញុំ។

https://github.com/DFoly/User_log_pipeline

អស់ហើយ។ អានភាគមួយ។.

ប្រភព: www.habr.com