こんにちは！オールアバウトでデータエンジニアをしている@ondaljhです。

「データエンジニアってどんなことやってるの？」という方々にも、こちらを読めばデータエンジニアの業務の一片がわかるようになる！連載企画「ログ物語」の第二回をお届けいたします。

バックナンバー

【第1章】ログ物語の始まり

今回の構成

第2章.ログ/収集の仲間

1.ログ収集
2.業務データ

第2章.ログ/収集の仲間

1.ログ収集

話す内容としては、All Aboutメディアを含む、PrimeAd(以下、PA)のメディアパートナーサイトからTreasure Data(以下、TD)までどの経由でログが収集されるかが対象になります。

ログ収集の概要

前回の記事で既にお見せしたことがある図になりますが、ログ収集の概要を簡略に表すと下記の図の通りになります。

f:id:allabout-techblog:20200218111541p:plain — ログの収集

ログの転送と集約・フィルタは両方ともfluentdを使っていますが、転送(いわゆるFowarder)サーバーと集約・フィルタ(いわゆるAggregatorとProcessor)サーバーをそれぞれ用意し、各自の役割を担っています。

ログ転送(Fowarder)

All Aboutメディアを含む、PAのメディアパートナーサイトからのログは各サイトに仕込んでるイメージトラッキングやリダイレクターによってログ転送サーバーに転送されます。ログ転送サーバーに転送されたログは、fluentdエージェントによって、ログ集約・フィルタサーバーにアクセスログとして転送されます。基本的にサーバーのスケールアウトが必要な時以外は設定変更が発生しません。ログ転送サーバーは、各サーバーから発生したログをログ集約・フィルタサーバーに右から左へ流すだけなので、サーバーインスタンスはAWSの汎用インスタンスを複数使っています。

ログ集約・フィルタ(AggregatorとProcessor)

ログ集約・フィルタサーバーではアクセスログをfluentdの機能を使ってフィルタリングやデータの簡単な正規化等を行ってからTDまで転送します。新しいログパターンが発生したり、新しいフィルタを適用したい場合には設定変更が必要になります。ログ転送サーバーとは異なり、一つ一つのログに対してfluentdの処理を行っているため、サーバーインスタンスはAWSのコンピューティング最適化インスタンスを複数使っています。

上記のように、ログ転送サーバーとログ集約・フィルタサーバーは役割も異なり、設定変更の発生頻度も異なります。また、役割によってサーバーへの負荷も異なります。従って、PrimeAdでは下記の観点でログ収集サーバーを各役割ごとに分けています。

サーバーの役割の明確さ
設定変更の頻度による障害リスクの軽減 (設定変更が必要なサーバーのみ設定変更を行う)
負荷に対するメンテナンスのしやすさ (負荷が高まった役割のサーバーをスケールアウトする)

ログ転送とログ集約・フィルタの構成図

ここまで見てきたログ転送サーバーとログ集約・フィルタサーバーを図で表すと、下記のようになります。

f:id:allabout-techblog:20200225110817p:plain — fluentdサーバー構成

上の図では、TD以外にもBigQueryが登場しています。BigQueryは下記の役割を担っています。

データウェアハウスとして、広告配信の機械学習や目標クリック到達判定で利用するためのリアルタイム集計用の生ログを格納。

学習やリアルタイム集計等は別の機会があれば説明することにしますので、この場では「そのようなものがあるんだ～」程度で問題ありません。

fluentdとは？

Fluentd is an open source data collector for unified logging layer. 一言で言うと、各種ログをデータとして収集するオープンソースになります。集計チームではログの転送及び集約・フィルタでもう既に何年も前から使っている状況です。 2018年12月時点での利用バージョンは下記になります。

ログ転送 : 0.12.12
ログ集約・フィルタ : 1.0.2
ログ集約・フィルタサーバーではマルチスレッドが使いたかったので、2018年から1.0.2を使っています。

簡単な説明

fluentdはRubyとCで実装されていて、Rubyのデーモンとして動きます。従って、設定ファイルもRuby処理ができるようになっています。そして、fluentdは数多くのプラグインが提供されていて、ログの収集でも下記のようなプラグインを使っています(プラグインの説明は省きますが…)。

fluent-config-regexp-type (1.0.0)
fluent-logger (0.7.1)
fluent-plugin-bigquery (1.2.0)
fluent-plugin-extract_query_params (0.1.1)
fluent-plugin-forest (0.3.3)
fluent-plugin-record-modifier (1.1.0)
fluent-plugin-record-reformer (0.9.1)
fluent-plugin-rewrite (0.1.1)
fluent-plugin-rewrite-tag-filter (2.1.0)
fluent-plugin-td (1.0.0)
fluent-plugin-td-monitoring (0.2.3)
fluentd (1.0.2)

fluentdは設定ファイルにディレクティブを書くことで設定できるし、設定した内容通りに動作します(これは当たり前ですねw)。基本的なディレクティブは下記になります。

ディレクティブ名	概要
`<source>~</source>`	入力プラグインの指定を行う
`<filter>~</filter>`	フィルタリングプラグインの指定を行う
`<match>~</match>`	出力プラグインの指定を行う

実際にTDへログを転送する処理では数多くのディレクティブ処理が行われています。下はそのソースの一部を抜粋したものになります。(一部マスキング済み)

#-------------------
# ログ転送サーバーからログの受け取り
#-------------------
<source>
  @type forward
</source>

#-------------------
# 不要行及びブラックリスト削除(共通処理)
# 共通処理の場合、ユーザエージェントは「ua」として渡される
#-------------------
<filter raw.**>
  @type grep
  <regexp>
    key uri
    pattern /hogehoge.gifuga.php
  </regexp>
  <exclude>
    key ua
    pattern bot|Bot|BOT
  </exclude>
</filter>

#-------------------
# uriから特定パラメータを取得する
#-------------------
<match raw.**>
  @type extract_query_params
  key uri
  remove_tag_prefix raw.log
  add_tag_prefix after.check.xxxtag
  only xxx, v
</match>

#-------------------
# xxxパラメータがあるかないかによって分岐
#-------------------
<match after.check.xxxtag>
  @type rewrite_tag_filter

  # xxxパラメータがある場合 - 英数字のみOK
  <rule>
    key xxx
    pattern ^[0-9a-zA-Z]+$
    tag decoding.xxxtag
  </rule>
</match>

Fluentd v0.12 Filter プラグインの使い方と作り方等を参照したので、興味ある方は一読してみてください。

2.業務データ

集計処理を行うためには意図したログなのかどうか等を判定するためのデータが存在し、集計側ではそのデータを「集計としてのマスタ」という意味で、「マスタデータ」と呼んでいます。が、その正体は「業務データ」になりますので、ここでは「業務データ」として説明させていただきます。

※正規化については前回の記事で軽く書いていますが、「第3章.ログ/二つの塔」にて詳しく説明させていただきます。

業務データとは

上記でも少し話しましたが、各種ログが意図したログかどうかを判定する材料になります。
例えば、All Aboutの一つの商品であるタイアップ記事の場合、掲載期間がありますが、既に掲載期間が過ぎた場合でも記事として残しています。
しかし、掲載期間が過ぎたタイアップ記事に対しては、掲載期間が終わった時点でもう集計は不要になります。
そのようなタイアップ記事データを集計しないように、ある業務データと結合して、掲載期間内のログだけを集計する等の処理を行います(「第3章.ログ/二つの塔」で詳しく説明しますが、この処理も正規化の1つになります)。

業務データの処理フロー

集計としての業務データは各種フロントデータに依存しているため、各種フロント側からデータを取得します。
フロント側のデータをそのまま使うパターンもありますが、集計用途に合わせて加工してから使うパターンがもっとも多いです。
取得して加工まで終わったデータは、TDへアップロードされ、そこから実際のログデータと結合されて利用されます。
文言だけだとわかり辛いので、簡単な図で出しますと、下記の感じになります。

f:id:allabout-techblog:20200225114222p:plain — 業務データ取得及びアップロード

この処理はバッチサーバーからTDのEmbulk機能を使ってMySQLから直接TDへのアップロードを行うようになっています。
Embulkは大量データの転送を楽にしてくれる技術で、業務データのアップロードでは下記サイトを参考にしました。
Bulk Import for MySQL

まとめ

PAのメディアパートナーサイトから生成されたログをどうやって収集し、TDまで蓄積しているかを紹介させていただきました。また、ログデータが意図したログデータであるかどうかを判定する等に使ってる、業務データの処理についても簡略に説明させていただきました。次回はこのように収集されたログデータをBIツールや人が使えるようにするための処理である、正規化と集計について書いていきたいと思います。