ਅਸੀਂ ਇੱਕ ਸਟ੍ਰੀਮ ਡੇਟਾ ਪ੍ਰੋਸੈਸਿੰਗ ਪਾਈਪਲਾਈਨ ਬਣਾਉਂਦੇ ਹਾਂ। ਭਾਗ 2

ਸਾਰਿਆਂ ਨੂੰ ਸਤਿ ਸ਼੍ਰੀ ਅਕਾਲ. ਅਸੀਂ ਲੇਖ ਦੇ ਅੰਤਮ ਹਿੱਸੇ ਦਾ ਅਨੁਵਾਦ ਸਾਂਝਾ ਕਰ ਰਹੇ ਹਾਂ, ਖਾਸ ਤੌਰ 'ਤੇ ਕੋਰਸ ਦੇ ਵਿਦਿਆਰਥੀਆਂ ਲਈ ਤਿਆਰ ਕੀਤਾ ਗਿਆ ਹੈ। ਡਾਟਾ ਇੰਜੀਨੀਅਰ. ਤੁਸੀਂ ਪਹਿਲਾ ਭਾਗ ਪੜ੍ਹ ਸਕਦੇ ਹੋ ਇੱਥੇ.

ਰੀਅਲ-ਟਾਈਮ ਪਾਈਪਲਾਈਨਾਂ ਲਈ ਅਪਾਚੇ ਬੀਮ ਅਤੇ ਡੇਟਾਫਲੋ

Google ਕਲਾਊਡ ਸੈੱਟਅੱਪ ਕੀਤਾ ਜਾ ਰਿਹਾ ਹੈ

ਨੋਟ: ਮੈਂ ਪਾਈਪਲਾਈਨ ਨੂੰ ਚਲਾਉਣ ਅਤੇ ਕਸਟਮ ਲੌਗ ਡੇਟਾ ਨੂੰ ਪ੍ਰਕਾਸ਼ਿਤ ਕਰਨ ਲਈ ਗੂਗਲ ਕਲਾਉਡ ਸ਼ੈੱਲ ਦੀ ਵਰਤੋਂ ਕੀਤੀ ਕਿਉਂਕਿ ਮੈਨੂੰ ਪਾਈਥਨ 3 ਵਿੱਚ ਪਾਈਪਲਾਈਨ ਚਲਾਉਣ ਵਿੱਚ ਮੁਸ਼ਕਲ ਆ ਰਹੀ ਸੀ। ਗੂਗਲ ਕਲਾਉਡ ਸ਼ੈੱਲ ਪਾਈਥਨ 2 ਦੀ ਵਰਤੋਂ ਕਰਦਾ ਹੈ, ਜੋ ਕਿ ਅਪਾਚੇ ਬੀਮ ਨਾਲ ਵਧੇਰੇ ਅਨੁਕੂਲ ਹੈ।

ਪਾਈਪਲਾਈਨ ਸ਼ੁਰੂ ਕਰਨ ਲਈ, ਸਾਨੂੰ ਸੈਟਿੰਗਾਂ ਵਿੱਚ ਥੋੜਾ ਜਿਹਾ ਖੋਦਣ ਦੀ ਲੋੜ ਹੈ। ਤੁਹਾਡੇ ਵਿੱਚੋਂ ਜਿਨ੍ਹਾਂ ਨੇ ਪਹਿਲਾਂ GCP ਦੀ ਵਰਤੋਂ ਨਹੀਂ ਕੀਤੀ ਹੈ, ਤੁਹਾਨੂੰ ਇਸ ਵਿੱਚ ਦੱਸੇ ਗਏ ਹੇਠਾਂ ਦਿੱਤੇ 6 ਕਦਮਾਂ ਦੀ ਪਾਲਣਾ ਕਰਨ ਦੀ ਲੋੜ ਹੋਵੇਗੀ ਸਫ਼ਾ.

ਇਸ ਤੋਂ ਬਾਅਦ, ਸਾਨੂੰ ਆਪਣੀਆਂ ਸਕ੍ਰਿਪਟਾਂ ਨੂੰ ਗੂਗਲ ਕਲਾਉਡ ਸਟੋਰੇਜ 'ਤੇ ਅਪਲੋਡ ਕਰਨ ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਸਾਡੇ ਗੂਗਲ ਕਲਾਉਡ ਸ਼ੈਲ 'ਤੇ ਕਾਪੀ ਕਰਨ ਦੀ ਜ਼ਰੂਰਤ ਹੋਏਗੀ। ਕਲਾਉਡ ਸਟੋਰੇਜ 'ਤੇ ਅਪਲੋਡ ਕਰਨਾ ਬਹੁਤ ਮਾਮੂਲੀ ਹੈ (ਇੱਕ ਵੇਰਵਾ ਪਾਇਆ ਜਾ ਸਕਦਾ ਹੈ ਇੱਥੇ). ਸਾਡੀਆਂ ਫਾਈਲਾਂ ਦੀ ਨਕਲ ਕਰਨ ਲਈ, ਅਸੀਂ ਹੇਠਾਂ ਚਿੱਤਰ 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 ਵਿੱਚ ਇੱਕ ਡਾਟਾਸੈਟ ਅਤੇ ਸਾਰਣੀ ਬਣਾਉਣਾ। ਅਜਿਹਾ ਕਰਨ ਦੇ ਕਈ ਤਰੀਕੇ ਹਨ, ਪਰ ਸਭ ਤੋਂ ਸਰਲ ਹੈ ਪਹਿਲਾਂ ਡੇਟਾਸੈਟ ਬਣਾ ਕੇ ਗੂਗਲ ਕਲਾਉਡ ਕੰਸੋਲ ਦੀ ਵਰਤੋਂ ਕਰਨਾ। ਤੁਸੀਂ ਹੇਠਾਂ ਦਿੱਤੇ ਕਦਮਾਂ ਦੀ ਪਾਲਣਾ ਕਰ ਸਕਦੇ ਹੋ ਲਿੰਕ ਨੂੰਇੱਕ ਸਕੀਮਾ ਨਾਲ ਇੱਕ ਸਾਰਣੀ ਬਣਾਉਣ ਲਈ. ਸਾਡੀ ਮੇਜ਼ ਹੋਵੇਗੀ 7 ਕਾਲਮ, ਹਰੇਕ ਉਪਭੋਗਤਾ ਲੌਗ ਦੇ ਭਾਗਾਂ ਦੇ ਅਨੁਸਾਰੀ। ਸਹੂਲਤ ਲਈ, ਅਸੀਂ ਸਮਾਂ-ਸਥਾਨਕ ਵੇਰੀਏਬਲ ਨੂੰ ਛੱਡ ਕੇ, ਸਾਰੇ ਕਾਲਮਾਂ ਨੂੰ ਸਤਰ ਵਜੋਂ ਪਰਿਭਾਸ਼ਿਤ ਕਰਾਂਗੇ, ਅਤੇ ਉਹਨਾਂ ਨੂੰ ਉਹਨਾਂ ਵੇਰੀਏਬਲਾਂ ਦੇ ਅਨੁਸਾਰ ਨਾਮ ਦੇਵਾਂਗੇ ਜੋ ਅਸੀਂ ਪਹਿਲਾਂ ਤਿਆਰ ਕੀਤੇ ਸਨ। ਸਾਡੀ ਸਾਰਣੀ ਦਾ ਖਾਕਾ ਚਿੱਤਰ 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ਇਸ ਨੂੰ ਪੂਰਾ ਕਰਨ ਲਈ.

ਚਿੱਤਰ 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 ਇਸ ਨੂੰ ਕੰਮ ਕਰਨ ਲਈ ਇੱਕ ਫੰਕਸ਼ਨ ਦੇ ਅੰਦਰ. ਮੈਨੂੰ ਫਾਈਲ ਦੀ ਸ਼ੁਰੂਆਤ ਵਿੱਚ ਇੱਕ ਆਯਾਤ ਗਲਤੀ ਮਿਲ ਰਹੀ ਸੀ, ਜੋ ਕਿ ਅਜੀਬ ਸੀ. ਇਹ ਸੂਚੀ ਫਿਰ ਫੰਕਸ਼ਨ ਨੂੰ ਪਾਸ ਕੀਤੀ ਜਾਂਦੀ ਹੈ ਰਾਈਟ ਟੂ ਬਿਗਕੁਆਰੀ, ਜੋ ਸਿਰਫ਼ ਸਾਡੇ ਡੇਟਾ ਨੂੰ ਸਾਰਣੀ ਵਿੱਚ ਜੋੜਦਾ ਹੈ। ਬੈਚ ਡੇਟਾਫਲੋ ਜੌਬ ਅਤੇ ਸਟ੍ਰੀਮਿੰਗ ਡੇਟਾਫਲੋ ਜੌਬ ਲਈ ਕੋਡ ਹੇਠਾਂ ਦਿੱਤਾ ਗਿਆ ਹੈ। ਬੈਚ ਅਤੇ ਸਟ੍ਰੀਮਿੰਗ ਕੋਡ ਵਿੱਚ ਸਿਰਫ ਫਰਕ ਇਹ ਹੈ ਕਿ ਬੈਚ ਵਿੱਚ ਅਸੀਂ 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 ਇੱਕ ਪਾਈਪਲਾਈਨ ਦੌੜਾਕ ਹੈ ਜੋ ਤੁਹਾਡੇ ਪ੍ਰੋਗਰਾਮ ਦਾ ਵਿਸ਼ਲੇਸ਼ਣ ਕਰੇਗਾ ਅਤੇ ਤੁਹਾਡੀ ਪਾਈਪਲਾਈਨ ਦਾ ਨਿਰਮਾਣ ਕਰੇਗਾ। ਕਲਾਉਡ ਵਿੱਚ ਚਲਾਉਣ ਲਈ, ਤੁਹਾਨੂੰ ਇੱਕ 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

ਸਿੱਟਾ

ਅਸੀਂ ਆਸ ਕਰਦੇ ਹਾਂ ਕਿ ਇਹ ਪੋਸਟ ਇੱਕ ਸਟ੍ਰੀਮਿੰਗ ਡੇਟਾ ਪਾਈਪਲਾਈਨ ਬਣਾਉਣ ਦੇ ਨਾਲ-ਨਾਲ ਡੇਟਾ ਨੂੰ ਵਧੇਰੇ ਪਹੁੰਚਯੋਗ ਬਣਾਉਣ ਦੇ ਤਰੀਕੇ ਲੱਭਣ ਦੀ ਇੱਕ ਉਪਯੋਗੀ ਉਦਾਹਰਣ ਵਜੋਂ ਕੰਮ ਕਰੇਗੀ। ਇਸ ਫਾਰਮੈਟ ਵਿੱਚ ਡਾਟਾ ਸਟੋਰ ਕਰਨ ਨਾਲ ਸਾਨੂੰ ਬਹੁਤ ਸਾਰੇ ਫਾਇਦੇ ਮਿਲਦੇ ਹਨ। ਹੁਣ ਅਸੀਂ ਮਹੱਤਵਪੂਰਨ ਸਵਾਲਾਂ ਦੇ ਜਵਾਬ ਦੇਣਾ ਸ਼ੁਰੂ ਕਰ ਸਕਦੇ ਹਾਂ ਜਿਵੇਂ ਕਿ ਕਿੰਨੇ ਲੋਕ ਸਾਡੇ ਉਤਪਾਦ ਦੀ ਵਰਤੋਂ ਕਰਦੇ ਹਨ? ਕੀ ਤੁਹਾਡਾ ਉਪਭੋਗਤਾ ਅਧਾਰ ਸਮੇਂ ਦੇ ਨਾਲ ਵਧ ਰਿਹਾ ਹੈ? ਉਤਪਾਦ ਦੇ ਕਿਹੜੇ ਪਹਿਲੂਆਂ ਨਾਲ ਲੋਕ ਸਭ ਤੋਂ ਵੱਧ ਗੱਲਬਾਤ ਕਰਦੇ ਹਨ? ਅਤੇ ਕੀ ਅਜਿਹੀਆਂ ਗਲਤੀਆਂ ਹਨ ਜਿੱਥੇ ਨਹੀਂ ਹੋਣੀਆਂ ਚਾਹੀਦੀਆਂ ਹਨ? ਇਹ ਉਹ ਸਵਾਲ ਹਨ ਜੋ ਸੰਸਥਾ ਲਈ ਦਿਲਚਸਪੀ ਦੇ ਹੋਣਗੇ. ਇਹਨਾਂ ਸਵਾਲਾਂ ਦੇ ਜਵਾਬਾਂ ਤੋਂ ਉਭਰਨ ਵਾਲੀਆਂ ਸੂਝਾਂ ਦੇ ਆਧਾਰ 'ਤੇ, ਅਸੀਂ ਉਤਪਾਦ ਨੂੰ ਬਿਹਤਰ ਬਣਾ ਸਕਦੇ ਹਾਂ ਅਤੇ ਉਪਭੋਗਤਾ ਦੀ ਸ਼ਮੂਲੀਅਤ ਵਧਾ ਸਕਦੇ ਹਾਂ।

ਬੀਮ ਇਸ ਕਿਸਮ ਦੀ ਕਸਰਤ ਲਈ ਅਸਲ ਵਿੱਚ ਲਾਭਦਾਇਕ ਹੈ ਅਤੇ ਇਸ ਦੇ ਨਾਲ-ਨਾਲ ਕਈ ਹੋਰ ਦਿਲਚਸਪ ਵਰਤੋਂ ਦੇ ਕੇਸ ਵੀ ਹਨ। ਉਦਾਹਰਨ ਲਈ, ਤੁਸੀਂ ਰੀਅਲ ਟਾਈਮ ਵਿੱਚ ਸਟਾਕ ਟਿੱਕ ਡੇਟਾ ਦਾ ਵਿਸ਼ਲੇਸ਼ਣ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹੋ ਅਤੇ ਵਿਸ਼ਲੇਸ਼ਣ ਦੇ ਅਧਾਰ ਤੇ ਵਪਾਰ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹੋ, ਸ਼ਾਇਦ ਤੁਹਾਡੇ ਕੋਲ ਵਾਹਨਾਂ ਤੋਂ ਆਉਣ ਵਾਲੇ ਸੈਂਸਰ ਡੇਟਾ ਹਨ ਅਤੇ ਟ੍ਰੈਫਿਕ ਪੱਧਰ ਦੀ ਗਣਨਾ ਕਰਨਾ ਚਾਹੁੰਦੇ ਹੋ. ਤੁਸੀਂ, ਉਦਾਹਰਨ ਲਈ, ਇੱਕ ਗੇਮਿੰਗ ਕੰਪਨੀ ਵੀ ਹੋ ਸਕਦੇ ਹੋ ਜੋ ਉਪਭੋਗਤਾ ਡੇਟਾ ਇਕੱਠਾ ਕਰਦੀ ਹੈ ਅਤੇ ਮੁੱਖ ਮੈਟ੍ਰਿਕਸ ਨੂੰ ਟਰੈਕ ਕਰਨ ਲਈ ਡੈਸ਼ਬੋਰਡ ਬਣਾਉਣ ਲਈ ਇਸਦੀ ਵਰਤੋਂ ਕਰਦੀ ਹੈ। ਠੀਕ ਹੈ, ਸੱਜਣ, ਇਹ ਇੱਕ ਹੋਰ ਪੋਸਟ ਲਈ ਇੱਕ ਵਿਸ਼ਾ ਹੈ, ਪੜ੍ਹਨ ਲਈ ਧੰਨਵਾਦ, ਅਤੇ ਉਹਨਾਂ ਲਈ ਜੋ ਪੂਰਾ ਕੋਡ ਦੇਖਣਾ ਚਾਹੁੰਦੇ ਹਨ, ਹੇਠਾਂ ਮੇਰੇ GitHub ਦਾ ਲਿੰਕ ਹੈ।

https://github.com/DFoly/User_log_pipeline

ਇਹ ਸਭ ਹੈ. ਭਾਗ ਪਹਿਲਾ ਪੜ੍ਹੋ.

ਸਰੋਤ: www.habr.com