こんにちは。メディアプロダクト開発部の我妻謙樹（id:kenju）です。サーバーサイドエンジニアとして、広告配信システムの開発・運用を担当しています。

今回は、cookpad storeTV （以下略：storeTV ）の広告商品における、リアルタイムログ集計基盤の紹介をします。

storeTV における広告開発

storeTV とは？

storeTV は、スーパーで料理動画を流すサービスで、店頭に独自の Android 端末を設置し、その売り場に適したレシピ動画を再生するサービスです。

より詳しいサービス概要にについては、弊社メンバーの Cookpad TechConf 2018 における以下の発表スライドを御覧ください。

storeTV における広告商品の概要

storeTV では、imp 保証型の広告商品を提供予定です。imp 保証型の広告商品とは、例えば「週に N 回広告を表示することを保証する」商品です。storeTV では、レシピ動画を複数回表示させるたびに広告動画を再生する仕様を想定しています。

f:id:itiskj:20181017154217j:plain

基本的には、実際の再生稼働数から在庫（広告動画の再生できる総回数）をおおまかに予測し、その在庫を元に比率を計算、配信比率を含む配信計画を全国に設置されている端末に配布する流れになります。

具体的な計算ロジックについて説明します。

配信比率は１時間を単位として算出しています。ある 1 時間の「配信比率」は、「 1 時間あたりの割当量 / 1 時間あたりの在庫量」と定義しています。

f:id:itiskj:20181017154304j:plain

Ratio ... 「配信比率」

HourlyAllocated ... 「1 時間あたりの割当量」

HourlyInventory ... 「1 時間あたりの在庫量」

ここで、「1 時間あたりの割当量」とは、1 時間あたりに再生したい回数です。広告商品によって達成したい目標 imp が異なりますので、「目標 imp」と「実績 imp」の差を、「残り単位時間」で割ることで算出できます。

また、「1 時間あたりの在庫量」とは、１時間あたりに再生され得る最大の広告再生回数と定義しています。storeTV の imp 保証型の広告商品の場合、先述したとおり「90 秒に 1 回配信される」という特性から、「1 時間あたりの最大広告再生回数」は 40 回と置くことができます。

したがって、「1 時間あたりの在庫量」は、「現在稼働中の端末数 * １時間あたりの最大広告再生回数」で算出できます。

f:id:itiskj:20181017154316j:plain

Goal ... 「目標 imp」

TotalImpression ... 「実績 imp」

RemainingHours ... 「残り単位時間」

= MaxCommercialMoviePlayCount ... 「1 時間あたりの最大広告再生回数」

最後に、「現在稼働中の端末数」ですが、正確な値をリアルタイムで取ることはほぼ不可能であるため、実際に発生した imp ログから大体の稼働数を予測する必要があります。

具体的には、まだ imp ログの発生が見られていない時（例：広告配信開始直後で、まだどの端末からもログが到達していない状態）は、「配布済みの全端末数 / カテゴリ数¹」で算出したものをデフォルト値として利用します。

imp ログが到達し始めたら、「直近 2 時間あたりの imp / 1 時間あたりの最大広告再生回数 / 直近 2 時間あたりの配信比率」で大体の値を算出しています。

f:id:itiskj:20181017154328j:plain

TotalDeviceCount ... 「配布済みの全端末数」

CategoryCount ... 「カテゴリ数」

EstimatedRunningDeviceCount ... 「現在稼働中の端末数」

HourlyImpression(currenthour - 2hours) ... 「直近 2 時間あたりの imp」

MaxCommercialMovierPlaycount ... 「1 時間あたりの最大広告再生回数」

Ratio(currenthour - 2hours) ... 「直近 2 時間あたりの配信比率」

storeTV ならではの課題

これだけでみると、一見簡単そうな要件に見えますが、cookpad 本体で配信している純広告とは異なる課題が多々存在します。

例えば、実際のハードウェアの配布数と実際の稼働数が大きく乖離するという点です。具体的には、「システム起因」（ネットワーク遅延、アップデート不具合）や「オペレーション起因」（バッテリー切れ、動画を流す手順操作をしない）などの様々な理由によって、広告動画が再生できない状態にある可能性があります。そういった広告が再生できない端末数を加味して配信比率を計算しないと、商品としての保証再生数を達成できないリスクが生じてしまいます。

そこで、稼働状態を把握するために、動画再生のログをリアルタイムで集計する必要がありました。リアルタイムで imp ログを参照できることで、より実際の稼働状況に即した配信比率が計算できます。より正確な配信比率に沿って広告を再生できることで、imp がショートしてしまったり、逆に出すぎてしまったりするようなリスクを可能な限り避けることができます。

なお、本件については、以下の資料もご参考ください。

cookpad storeTV での広告配信のいままでとこれから

storeTV 広告におけるリアルタイムログ集計基盤

そこで、リアルタイムログ集計基盤を用意する必要がありました。次節より、具体的なアーキテクチャや、実装上の課題、それに対する工夫についてお話しておきます。adtech研究会#1 にて発表した以下のスライドがベースとなっています。

当プロジェクトをまさに実装中に発表した資料で、ストリーミング処理における理論や資料についてまとめてあります。そのため、今回紹介する最終形とは異なり、一部古い記述（具体的には、Analysis レイヤーのアーキテクチャ）があります。

Architecture

まずは、全体のアーキテクチャを紹介します。

f:id:itiskj:20181017154341j:plain

主に、以下の 3 レイヤーから構成されています。

Aggregation ... imp ログを集計するレイヤー
Monitoring ... ストリームの詰まりやログの遅延を検出するためのモニタリングレイヤー
Analysis ... リアルタイムログを、分析用に Redshift に入れるための分析レイヤー

先ほど説明したように、元々は配信比率の計算のためのリアルタイムでの imp ログの集約が最優先タスクでした。それに伴い、Aggregation/Monitoring レイヤーを作成しました。

その後、せっかく Kinesis Streams ですべての広告動画再生ログを受け取っているので、ニアリアルタイム（大体 10 ~ 15 分程度）で分析するための Analysis レイヤーを作成しました。弊社では、DWH チームが提供するデータ活用基盤に分析するデータを寄せることで、全社的に品質の高い分析フローを低コストで用意することができます。今回も、S3 Bucket に Kinesis Firehose 経由で投げたら、後は DWH の仕組みに乗って Redshift にデータが入ってきます。

Aggregation

f:id:itiskj:20181017154354j:plain

Core Design

Aggregation レイヤーでは、Android 端末から、以下のような JSON ログが送信されてきます。

{
  "delivery_id": 123,
  "delivery_plan_commercial_movie_hourly_goal_id": 100,
  "identifier": "foobar1234",
  "sending_time": "2018-07-12T10:22:58+09:00"
}

端末が送信してきたタイムスタンプに従って、1 分間ごとのログに集約させ、DynamoDB に格納します。

name	type	schema	example
delivery_id	String	Hash Key	`123`
period_id	Int	Sort Key	`201807121022`
impressions	Int	-	`1`

この時、DynamodDB - Update Expressions の ADD 演算を利用し、impressions の項目は、インクリメントさせます。これによって、JSON ログが流れてくるたびに、imp が徐々に増えていくようになります。

次に、このレコードを集約させた DynamoDB の Streams を有効にし、別の Lambda で吸い取ります。その Lambda は、minute 単位で格納されたレコードを、今度は hourly で集約し、別の DynamoDB に格納します。さらにその DynamoDB Streams から、daily 別に格納します。

このように徐々に集約させていくことで、ストリームデータを適切な粒度でカウントさせます。配信比率を計算するクライアントは、適宜必要なテーブルの imp レコードを参照します。

Late Logs & Watermarks

1 つめの Lambda が書き込んでいる DynamoDB が、2 つあることに気づいた方もいらっしゃるかもしれません。

f:id:itiskj:20181017154412j:plain

ここでは、ある一定時間以上遅延したログは、それ以降の集約ロジックに含めず、別の遅延専用テーブルに書き込んでいます。

一般に、ストリーミング処理において「遅延したログをどう扱うか」というのは、古くからある命題の一つです。ログは、様々な理由で遅延します。クライアントがオフライン状態にあったり、message broker が障害で落ちていたり、message broker からデータを吸い取り処理する consumer 側にバグが有ってデータが処理されなかったり、バグはないが想定以上のデータが来ることで consumer が一定時間以内に処理しきれず、かつスケールアウトも間に合わなかったり。

ストリーミング処理における理論や一般的プラクティスについては、『Designing Data-Intensive Applications』の十一章と、Oreilly における二部長編エッセイに詳しいので、興味がある方はそちらをおすすめします。

今回、遅延してきたログの対処法として、具体的には、Watermarks² で処理することにしました。Watermarks とは、「どこまでログを処理したか」という情報を「透かし」としてストリームデータに仕込み、各 consumer がデータを処理する時、遅延したかどうかを判別できるようにするものです。

特に、今回は Apache Spark における Watermarks 実装を参考に、オンメモリで計算できる設計にしました。外部データソースに、consumer ごとに「最後に処理したタイムスタンプ」を保存しておく実装も考えられましたが、別にテーブルを用意するコストや、毎回問い合わせが発生することによるコストを考慮し、避けました。

具体的には、冒頭で紹介したスライド p31 に擬似コードを載せています。「consumer が処理する一連のログのタイムスタンプの内、中央値に対してプラスマイナス 5 分以上乖離しているかどうか」で判別しています。

Apache Spark の例では最大値を利用していますが、storeTV ではクライアント側のシステム時間を操作できてしまうという特性上、かなり未来の時間が誤って送られてくることも考えられます。最大値を用いる手法だと、そのような外れ値に引っ張られて、本来集約ロジックに含めたい大量の正しいログを捨ててしまうことになります。ですので、中央値を利用しています。

Monitoring

Streaming Delay

次は、モニタリングです。ストリーミング処理においては、先程も説明したとおり、以下のような様々な理由によって詰まる可能性があります。

クライアントがオフライン状態
message broker が障害で落ちている
message broker からデータを吸い取り処理する consumer 側にバグが有ってデータが処理されない
バグはないが想定以上のデータが来ることで consumer が一定時間以内に処理しきれず、かつスケールアウトも間に合わない

f:id:itiskj:20181017154430j:plain

そこで、2 通りの方法を使ってモニタリングしています。

Kinesis Streams の GetRecords.IteratorAgeMilliseconds ³
DynamoDB に実際に書き込まれた imp ログのタイムスタンプが一定の閾値以内かどうか

まず、Kinesis Streams の IteratorAgeMilliseconds は、最後に処理されたレコードの時間を示します。もし、IteratorAge と現在時刻が乖離すればするほど、この値は大きくなっていきます。この値をモニタリングすることによって、Lambda 側でバグや障害が発生するなどしてレコードが処理されない事象を検知します。

また、DynamoDB に実際に書き込まれた imp ログが、現在時刻から一定の閾値以内かどうかを検知するための Lambda も存在しています。これは、例えば Kinesis に障害が発生するなどして、一定時間ストリーム自体が流れて来ず、復旧後に溜まっていた一連のストリームが流れてくるような場合を検知します。

CloudWatch Alarms の通知先として、SNS Topics を指定します。この Topics を、Slack に通知する責務をもたせた Lambda の event source として指定しておきます。こうすることで、通知ロジックの責務自体は分散させずに一連のモニタリングフローを整備できました。

CloudWatch Dashboard

また、Kinesis Streams や Lambda, DynamoDB など各種コンポーネントの稼働率も別途モニタリングする必要があります。今回は、全て AWS のサービス群で構成しているので、モニタリングツールには CloudWatch Dashboard を利用しました。既存の Metrics から必要な指標を手軽に作ることができるのでおすすめです。

Analysis

最後に、Analysis レイヤーです。Kinesis Streams の consumer として、Kinesis Firehose を指定し、S3 に吐くだけです。指定のフォーマットで S3 に出力した後は、DWH の仕組みに乗って Redshift に書き込まれていきます。

f:id:itiskj:20181017154511j:plain

ポイントは、Kinesis Firehose の Transformer として Lambda を指定している箇所です。Firehose では、Streams から来たログをそのまま S3 に流すのではなく、Lambda で前処理を加えることができます。これは、一般的な ETL⁴ 処理における、Transform 機構をシンプルに実現できるということです。

Firehose はバッファリングとしての機構を持つので、例えば「5分間 or ログの総量が 1 MB を越えたら S3 に出力する」といった設定をすることができます。No Configuration が Firehose のウリでもあるので、例えばこのバッファリングの頻度を変更したいときは、コードを変更することなく、AWS Console から設定一つで変更することができます。