ストリームデータ処理パイプラインを作成します。 パート2

こんにちは、みんな。 コースの学生向けに特別に用意された、記事の最後の部分の翻訳を共有します。 データエンジニア。 最初の部分を読むことができます ここで.

リアルタイム パイプラインのための Apache Beam と DataFlow

Google Cloud のセットアップ

注: 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])

コードでは XNUMX つのカスタム関数を作成します。 関数 regex_clean、データをスキャンし、関数を使用して PATTERNS リストに基づいて対応する行を取得します。 re.search。 この関数はカンマ区切りの文字列を返します。 正規表現の専門家でない場合は、これを確認することをお勧めします チュートリアル メモ帳で練習してコードを確認してください。 この後、カスタム ParDo 関数を定義します。 スプリットこれは、並列処理のための Beam 変換のバリエーションです。 Python では、これは特別な方法で行われます。DoFn Beam クラスを継承するクラスを作成する必要があります。 Split 関数は、前の関数から解析された行を取得し、BigQuery テーブルの列名に対応するキーを持つ辞書のリストを返します。 この関数について注意すべき点があります。インポートする必要がありました。 datetime それを機能させるために関数内で。 ファイルの先頭でインポート エラーが発生していましたが、これは奇妙でした。 このリストは関数に渡されます。 BigQuery に書き込むこれは単にデータをテーブルに追加するだけです。 バッチ DataFlow ジョブとストリーミング DataFlow ジョブのコードを以下に示します。 バッチ コードとストリーミング コードの唯一の違いは、バッチでは 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

それだけです。 最初の部分を読む.

出所： habr.com