스트림 데이터 처리 파이프라인을 생성합니다. 2 부

안녕하세요 여러분. 우리는 이 과정의 학생들을 위해 특별히 준비된 기사의 마지막 부분의 번역을 공유하고 있습니다. 데이터 엔지니어. 첫 번째 부분을 읽을 수 있습니다 여기에.

실시간 파이프라인을 위한 Apache Beam 및 DataFlow

Google 클라우드 설정

참고: Python 3에서 파이프라인을 실행하는 데 문제가 있었기 때문에 Google Cloud Shell을 사용하여 파이프라인을 실행하고 커스텀 로그 데이터를 게시했습니다. Google Cloud Shell은 Apache Beam과 더 일관된 Python 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열, 각 사용자 로그의 구성 요소에 해당합니다. 편의상 timelocal 변수를 제외한 모든 열을 문자열로 정의하고 앞서 생성한 변수에 따라 이름을 지정하겠습니다. 테이블의 레이아웃은 그림 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 파이프라인을 생성하려면 파이프라인 객체(p)를 생성해야 합니다. 파이프라인 개체를 생성한 후에는 연산자를 사용하여 여러 기능을 차례로 적용할 수 있습니다. pipe (|). 일반적으로 워크플로는 아래 이미지와 같습니다.

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

우리 코드에서는 두 개의 사용자 정의 함수를 생성할 것입니다. 기능 regex_clean데이터를 스캔하고 함수를 사용하여 PATTERNS 목록을 기반으로 해당 행을 검색합니다. re.search. 이 함수는 쉼표로 구분된 문자열을 반환합니다. 정규식 전문가가 아닌 경우 이 내용을 확인하는 것이 좋습니다. 지도 시간 코드를 확인하려면 메모장에서 연습하세요. 그 후에는 다음과 같은 사용자 정의 ParDo 함수를 정의합니다. 스플릿이는 병렬 처리를 위한 빔 변환의 변형입니다. Python에서는 이 작업이 특별한 방식으로 수행됩니다. 즉, DoFn Beam 클래스에서 상속되는 클래스를 만들어야 합니다. Split 함수는 이전 함수에서 파싱된 행을 가져와 BigQuery 테이블의 열 이름에 해당하는 키가 포함된 사전 목록을 반환합니다. 이 함수에 대해 주의할 점이 있습니다. datetime 함수 내부에서 작동하도록 합니다. 파일 시작 부분에서 가져오기 오류가 발생했는데, 이상했습니다. 그런 다음 이 목록은 함수에 전달됩니다. WriteToBigQuery, 이는 단순히 데이터를 테이블에 추가합니다. Batch DataFlow Job 및 Streaming DataFlow Job에 대한 코드는 다음과 같습니다. 일괄 코드와 스트리밍 코드의 유일한 차이점은 일괄적으로 다음에서 CSV를 읽는 것입니다. src_path기능을 사용하여 ReadFromText 빔에서.

일괄 DataFlow 작업(일괄 처리)

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 프로젝트의 ID입니다.
• runner 프로그램을 분석하고 파이프라인을 구성하는 파이프라인 실행기입니다. 클라우드에서 실행하려면 DataflowRunner를 지정해야 합니다.
• staging_location — 작업을 수행하는 프로세서에 필요한 코드 패키지 색인을 생성하기 위한 Cloud Dataflow 클라우드 저장소의 경로입니다.
• temp_location — 파이프라인이 실행되는 동안 생성된 임시 작업 파일을 저장하기 위한 Cloud Dataflow 클라우드 저장소의 경로입니다.
• streaming

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

이 명령이 실행되는 동안 Google 콘솔의 DataFlow 탭으로 이동하여 파이프라인을 볼 수 있습니다. 파이프라인을 클릭하면 그림 4와 비슷한 내용이 표시됩니다. 디버깅 목적으로 로그로 이동한 다음 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

그게 다야. XNUMX부 읽기.

출처 : habr.com