こんにちは。去年の今頃は Rust を書いていました。 インフラストラクチャー部データ基盤グループの id:koba789 です。
背景
クックパッドではデータ基盤の DBMS として Amazon Redshift を利用しています。 既存のデータ基盤について詳しいことは クックパッドのデータ活用基盤 - クックパッド開発者ブログ を参照してください。
今まで、ログは数時間に1度、定期実行ジョブで Redshift 内のテーブルにロードしていました。 ロードジョブの実行間隔が "数時間" と長めなのは、Redshift のトランザクションのコミットが遅いためです。 クックパッドでは数百ものログテーブルがあるため、仮に1分おきにすべてを取り込もうとすると秒間数回以上のコミットを行わなければなりません。 このような頻繁なコミットは Redshift 全体のパフォーマンスを悪化させてしまいます。
サービスの開発者はリリースした新機能の様子をすぐに確認したいものです。 にもかかわらず、ログがクエリできるようになるまで、最大で数時間も待たなければなりませんでした。 最悪の場合、その日のうちに確認することは叶わず、翌朝まで待たされることもありました。 これではサービスの改善を鈍化させてしまいます。
また、問題はもう一つありました。 当然ですが、ログは日々単調増加し、ストレージの容量を消費し続けます。 一方で、Redshift のストレージサイズはノード数に対して固定です。 ノードを追加すればストレージを増やせますが、ノードには CPU やメモリも付いているため、ストレージが欲しいだけの場合は割高です。 すなわち、ログにとって、Redshift のストレージは高価だったのです。
求められているもの
絶え間なく流れ続けるログを遅延なくロードし続けられるログ基盤が必要でした。 もちろん、ロードされたデータはなるべくすぐにクエリ可能になるべきです。
その上で、クエリの実行速度を犠牲にすることはできません。 クエリが遅くなれば夜間のバッチジョブの実行時間に影響が出ます。 そのため、従来の Redshift 内のテーブルへのクエリと同等かそれ以上のクエリ速度が求められていました。
そして最後に、願わくばストレージが安くて無限にスケールすることを。
この要件だけ見ると BigQuery を使えばいいのではないかと思われるかもしれません。 確かに BigQuery なら上記の要件は満たせるかもしれません。 しかし、Redshift にある PostgreSQL 互換の接続インターフェイスや AWS IAM との連携などの機能は BigQuery にはないため、Redshift を使うことで既に達成できているその他の要件が満たせなくなってしまいます。 また、大量のデータをクラウドプロバイダ間で日々転送しつづけることによって発生する追加の転送料も問題になります。
以上のような理由から、BigQuery に乗り換えるだけでは我々の理想は達成されないと判断しました。
Redshift Spectrum
まずはじめに、最強のログ基盤の一翼を担っている大切なコンポーネントである Redshift Spectrum を紹介します。
Redshift Spectrum は Redshift の機能のひとつです。 Redshift Spectrum を用いると、Redshift 内から S3 に置かれたデータを直接クエリすることができます。 にもかかわらず、Redshift 内のテーブルと JOIN することもできます。 つまり、S3 に置かれたデータを通常のテーブルと同様に扱うことができるのです。
Redshift Spectrum で読めるテーブル(以下、外部テーブル)の内容は Redshift へではなく、S3 に直接書き込みます。 そのため、トランザクションによる保護はなく、同一トランザクション内であっても、複数回の読み取りはそれぞれ別の結果を返す可能性があります。
しかし、今回はこの特性を逆手にとり、頻繁なロードを可能にしました。 Redshift の内部テーブルへの書き込みと違い、S3 への書き込みならコミットのパフォーマンスに悩まされることはない、という算段です。
そしてみなさんご存知のとおり、S3 は容量が事実上無限にスケールして容量単価も安価なストレージですので、最後の願いが叶えられることは言うまでもありません。
高速なクエリのために
Redshift Spectrum を活用するにあたって、高速なクエリを実現するために実践せねばならないプラクティスがいくつかあります。 これらのプラクティスは AWS の開発者ガイドを参考にしています。
Amazon Redshift Spectrum クエリパフォーマンスの向上 - Amazon Redshift
まず1つ目はパーティション化です。
Redshift Spectrum ではパーティション化をしないとクエリのたびにテーブルの全データをスキャンすることになってしまいます。 今回のケースでは、ログレコードの発生時刻の日付でパーティションを切ることにしました。 パーティション化によって、データオブジェクトの key はスキーマ名・テーブル名のあとに、さらに日付で分割され、以下のようなレイアウトになります。
hoge_schema.nanika_table/dt=2018-08-09/001.parquet002.parquet- ...
dt=2018-08-10/008.parquet009.parquet- ....
2つ目は列指向フォーマットの利用です。
Redshift Spectrum では、CSV や改行区切りの JSON などの一般的な行指向のテキストファイルもクエリすることができますが、クエリをより効率的かつ高速にするには、列指向フォーマットの利用が有効です。 列指向フォーマットにも様々な種類がありますが、今回のケースでは Parquet を採用しました。
3つ目は、各データオブジェクトのサイズを64MB以上の均等なサイズに揃えることです。
あまりに小さなデータオブジェクトは I/O オーバーヘッドの割合を増加させたり、Parquet の圧縮率を低下させたりします。 また、データオブジェクトのサイズの偏りは分散処理の効率性に悪影響を及ぼします。
上記の記事には、以上の3つの他に、効率的なクエリの書き方についてのプラクティスも紹介されていますが、ログのロードではそれらは関係ないため、ファイルの配置に関する内容のみを取り上げています。
Overview
Prism は Redshift Spectrum にログをロードするために私が開発したソフトウェアです。 Prism は3つのコンポーネントからなります。
1つ目は Prism Stream です。 これは S3 に到着した JSON 形式のログオブジェクトを Parquet に変換するコンポーネントです。
2つ目は Prism Merge です。 これは Prism Stream が書き出した細切れのデータオブジェクトを適切なサイズに結合(マージ)するコンポーネントです。
3つ目は Prism Catalog です。 日付変更後に当日分のパーティションを作成したり、階層の切り替え(後述)を行ったりするコンポーネントです。
弊社のコンテナ基盤である Hako を用いてスポットインスタンス上にデプロイされています。 スポットインスタンスの急な停止に耐えるため、上記のコンポーネントのすべての処理は、途中で突然終了してもデータの欠落や不整合が発生しないように設計されています。 Prism は RDB の読み書きのみならず、S3 への書き込みなども行います。 そのため、データの整合性については単に DBMS のトランザクションに委ねるというわけにはいかず、ケースバイケースでのケアが必要になります。*1
このようにすべての処理を冪等ないしはアトミックに実装したことの嬉しい副作用として、ネットワークや S3 の不調などによって偶発的にエラーが起きても、リトライするだけで回復できるという点があります。 もっとも、勝手にリトライするので人間がそれを気にすることは稀ですが。
また、S3 に配置したログオブジェクトや各パーティションのメタデータの DB として PostgreSQL を用いていますが、ログオブジェクトの数に比例して行が増えてしまうテーブルはローテート可能な設計にするなど、スケーラビリティのための工夫をしています。
階層化されたパーティション
各パーティションは prefix を用いて S3 上で論理的に階層化されています。 階層は SMALL と MERGED に分かれており、Glue カタログにはパーティションごとに SMALL では prefix が、MERGED ではマニフェストファイルの key が登録されています。 つまり、Redshift Spectrum が読めるのはパーティションごとに SMALL 階層か MERGED 階層のどちらか一方のみです。 当日(最新)のパーティションは初期状態で SMALL 階層を参照しており、マージ処理後、Prism Catalog によって順次 MERGED 階層を参照するように切り替わっていきます。
SMALL 階層のデータは分単位で細切れになっているため、オブジェクトサイズが最適ではなかったり不揃いだったりしており、高速なクエリには向いていません。 一方 MERGED 階層のデータは適切なサイズに揃えられており、高速なクエリのためのプラクティスに沿っています。
これは、低レイテンシなロードが必要になる場面と、高速なクエリが必要になる場面の違いを踏まえた設計です。
- 直近(1日程度)のデータについては低レイテンシである必要があるが、クエリは低速でよい
- "直近のデータ" である時点で対象が小さいため、クエリの速度は問題にならない
- より古いデータについてはクエリは高速でなければならないが、ロードのレイテンシは高くてよい
- 「今さっき届いた昨日分のログ」をすぐに見たいということは稀である
SMALL 階層のように細切れのオブジェクトを並べるだけではクエリが低速になってしまいますが、適切なサイズに粒を揃えるために単にバッファリングしてしまってはレイテンシが大きくなってしまいます。 これは低レイテンシなロードと高速なクエリを両立するための設計です。
データの流れと階層切り替え
では当日のパーティションに書き込まれた当日分のデータの流れを追ってみましょう。
まず、到着したログオブジェクトは Prism Stream によって Parquet に変換され SMALL 階層に書き込まれます。 この時点で、Redshift から読むことが可能になります。
その裏で、SMALL 階層に書き込まれたログオブジェクトは定期的に Prism Merge によってマージされ、マージ後のデータオブジェクトは MERGED 階層へ書き込まれます。 ただしまだこの時点では Redshift から MERGED 階層のデータを読むことはできません。
日付が変わったあと、Prism Catalog は SMALL 階層にある分のデータがすべて MERGED 階層に揃ったことを確認します。 確認が取れると、Prism Catalog は MERGED 階層にあるオブジェクトの一覧をマニフェストファイルに書き出し、Glue カタログにその key を登録することで、階層の切り替えをします。 この処理が走ることによって初めて、パーティションの参照先が MERGED 階層に切り替わります。 もしも MERGED 階層に SMALL 階層と同じだけのデータが揃う前に切り替えてしまうと、それまで読めていたデータが一部減少することになってしまいます。
Out-of-Order Data
当日に到着した、当日分のデータの流れについて説明しましたが、現実にはログは大幅に遅れて届くこともあります。
例として、モバイルアプリの行動ログの場合を説明します。 ユーザーが通信状況の悪い環境でアプリを操作したとします。 すると、ログレコード自体はその場で生成されますが、ログを送出することに失敗します。 クックパッドのモバイルアプリではログの送出に Puree というライブラリを使っており*2、上記のように送出に失敗したログは一旦端末に保管され、次の送出のチャンスを待ちます。 ここでユーザーがアプリを終了させ、翌日になってから通信状況の良好な環境で再度起動したとします。 通信状況が回復したため、アプリは前日の行動ログを再送します。 するとログ基盤には到着時刻に対して発生時刻が1日前になっているレコードが到着します。 ほかにも様々な理由により、ログのレコードはバラバラの順序で到着します。
そのため、日付が変わろうと当該日の SMALL 階層への書き込みが止むことはありません。 つまり、Prism Catalog が確認をしているその最中にも Prism Stream が新たなログオブジェクトを SMALL 階層に書き込むかもしれないということです。 これでは Prism Catalog が "SMALL 階層にある分のデータがすべて MERGED 階層に揃ったことを確認" することができません。
この問題を解決するため、Prism では「締め」という概念を導入し、SMALL 階層を「締め」前に書き込む LIVE 階層と「締め」後に書き込む DELAYED 階層に分割しました。 そして、SMALL 階層のうち、Redshift Spectrum が読み取り可能な部分は LIVE 階層のみとしました。
hoge_schema.nanika_table/live/(LIVE 階層)dt=2018-08-09/001.parquet002.parquet- ...
deleyed/(DELAYED 階層)dt=2018-08-09/010.parquet011.parquet- ...
merged/(MERGED 階層)dt=2018-08-09/001-007.parquet008-013.parquet- ...
「締め」の後では LIVE 階層のデータが増えないことが保証されるため、Prism Catalog が階層切り替えの判断をする際には、LIVE 階層のデータがすべて MERGED 階層に揃っているかどうかを安心して確認することができます。 LIVE 階層のデータが MERGED 階層に揃った後では、どのタイミングで切り替えてもデータの減少は起きませんので、LIVE 階層と MERGED 階層の比較だけもって階層切り替えをすることができます。
まとめ
最強のログ基盤を手に入れるために開発したソフトウェアと、その設計についてご紹介しました。 Redshift Spectrum のようなクラウドサービスはとても大きくて複雑なコンポーネントですが、それ自体の理解はもちろんのこと、自分たちの課題をじっと見つめ、ひとつひとつ丁寧にトレードオフを選択していくことで強力な武器となります。
Prism の各処理をいかにリトライ可能にしたかなど、まだまだ 自慢 解説したい内容は尽きないのですが、それについて書き始めるといつまで経っても本記事を公開できそうになかったため、ここで筆を置かせていただきました。
クックパッドでは絶対ジョブをリトライ可能にしたいエンジニアやデカいデータなんとかしたいエンジニアを募集しています。 Prism の設計や実装について興味があるという方はぜひともご応募ください。
*1:基本方針としては、なんらかアトミックな値の書き換えによってコミットとすることで Atomicity を作り込むというパターンですが、ここに記すには余白が狭すぎる
