こんにちは。プロダクト開発グループの福角です。 大量データを加工してBQに格納する場合の設計方針についてまとめました。
前提
- 元データの性質や管理方法によって、最適なプログラムの実装方法は違うと思いますので、この記事では、プログラム設計については言及しません。基本設計とその際の注意点について言及します。
- 今回、元データはRDSに格納されているケースを事例に出しますが、S3においてあるログ情報など、元データの格納場所がどこなのかは基本的に影響しません。
背景
弊社では、自社サービスを開発・運営しています。サービスを運営していく上で、RDBに格納された過去の情報（およそ300億ほどあるレコード情報）を加工して、BigQueryに保存する需要が生まれました。
実装方法
概要
BigQueryまでのデータの流れは以下のようになります。
BigQueryにデータを入れる場合、大きく分けて２つの方法があります。
- 読み込みジョブ: GCSなどに置いてあるファイルを一括で読み込んでBQにロードする。
- ストリーミングインサート: リアルタイムに1件ずつデータをBQにインサートする。
今回のようなリアルタイム性が必要なく、大量にデータを入れたい場合は前者の読み込みジョブを使うことが推奨されています（読み込みジョブは無料だが、ストリーミングインサートは有料）。
GCSに正しいフォーマット、ファイルパスでファイルをPUTできてしまえれば、BigQueryへのロードは容易ですので、GCSへ格納するところに焦点をあてます。
GCSのファイルパス設計
まず、GCSに置くファイルのパス設計について考えます。ポイントは以下。
BigQueryでは、「ディレクトリ下（厳密には単なるprefix）の全てのファイルをBQの指定テーブルにロードする」ことが可能です。
また、BigQueryには現状、日付パーティションが可能であり、レコードの発生日時をもとにパーティションします。
そして、パーティションが違うとロード先のテーブルが違うことを意味します。（例えば、2020年1月1日のデータは、<テーブル名>$20200101に格納することとなる）したがって、レコードの日付毎にファイルを格納するディレクトリを変える必要があります。
また、弊社はBtoBに属する業態であり、各クライアントが我々のサービスにアクセスしてを利用していただくものとなっております。
そのため、クライアント毎にテーブルを分けてデータ管理したいと考えております。（スキーマは同じだが、権限などでクライアント毎に柔軟に設定できるため）
上記より、以下のファイルパスでGCSに保存することになります。
gs://<バケット名>/YYYY/MM/DD/<クライアント名(テーブル名)>/(任意のファイル名).json.gz
一般的には、テーブル・パーティション毎にディレクトリを分けて保存することとなります。
注意点
今回のような大量データを取り扱うにおいて、重要な視点がコストです。 普段、気にしないような小さなコストポイントも、大量データの取り扱いとなると、料金が莫大になってきたりします。
私も入社したての際、今回とは取り扱うデータは違いますが、上記のように大量データをGCSに格納する仕事を任されました。
その際、コストの注意が足りず、ものすごい額になってしまった失敗経験があります。
今回の場合、特に注意する必要があるコストポイントは、GCS (Cloud Storage) のPUT料金になります。執筆時点では、10,000オペレーションあたり$0.05となっております。 https://cloud.google.com/storage/pricing?hl=ja#operations-pricing
仮に300億のレコードに対し、1レコード1ファイルでPUTした場合、15万ドル（1500万円以上）もかかってしまいます。
したがって、どれだけファイルを少なくしてPUTできるかが重要な課題となります。
コスト見積もり
ここでは、S3のPUTオペレーション料金に絞って見積もります。
前述のファイルパス設計の通り、(テーブル数) * (パーティション数)分のファイルPUTは最低限必要となります。
今回のケースの場合、2018年〜2020年のデータを格納するので、およそ1,000日分、つまり1000パーティション分（パーティション数には上限があるため、実際にはテーブル名に年情報を入れている）。
クライアントの数は、高々200程度となりますので、 200 (テーブル) * 1,000 (パーティション) = 200,000 PUTは最低限必要となります。
ただし、ロードジョブでは、１ファイルあたりの最大サイズには上限があるので、いくつかに分割しなければならないこともあると思います。
200,000 PUTであれば、100円程度となりますので、仮に1テーブル1パーティション毎に10ファイルくらいPUTしたとしても、許容範囲内でしょう。
GCSオブジェクト生成プログラム
前提で述べた通り、データの性質や管理方法によって、プログラム設計は変わってくるかと思いますので、具体的なコード設計は割愛します。ただし、今回、注意点としてあげたように、GCSへのPUT回数は極力減らす工夫が必要です。
上記の要件を満たすため、以下の方針を採用するケースもあるかと思います。
- A. 大量ファイルを一時的にローカル保存し、一定量処理したら、ファイル結合してGCSへアップロード。
- B. GCS(S3でも同じ)にはファイル追記機能はない。したがって、ローカルで、テーブル・パーティション毎にファイルを生成し、ファイル追記していく。一定量処理したら、GCSへアップロード。
c5d系インスタンスの採用を推奨
上記どちらの方法を採用しても、共通してローカルに大量の書き込みが発生します。
その際、オススメしたいのが、EC2インスタンスに c5d系インスタンスを採用することです。（ただし、元データがGCPにあるのであれば、わざわざAWSを使う必要はないと思います）
近頃では、サーバレスが流行ってきており、EC2インスタンスのブロックストレージの種類とかあまり気にしなくなってきたので忘れてたのですが、EC2には、２つのストレージタイプが存在します。EBSとインスタンスストアです。
多くのEC2インスタンスタイプでは、EBSのみが対応していますが、一部のEC2インスタンスタイプでは、インスタンスストアが利用できます。
その内の１つがc5d系インスタンスです。c系なのでコンピューティング性能を重視したインスタンスでありつつ、インスタンスストアが使えるタイプになります。
インスタンスストアとEBSの違いなどは他に優良記事がたくさん存在するので、そちらを参考にしていただければ良いと思いますが、概要だけ説明すると、インスタンスストアは揮発性（停止するとデータが消える）があるが、そのトレードオフとして、高パフォーマンスが期待できます。
今回のように一時データとして、保存するファイルなので、揮発性があることによる問題は特にありません。 また、書き込みが大量に発生するため、EBSではパフォーマンスに限界があり、すぐにバーストクレジットを使い切ってしまい、性能が劣化してしまいます。（当時はEBSのクレジット枯渇に気づかず苦労した）
その他プログラム設計する上で重要なこと
コード設計する上で、どんなプログラムコードにするにしても、以下の観点を考慮してプログラム設計しましょう。
- 数分で終わるようなプログラムではありません。途中でプログラムを終わりにしても、途中から再開できるようにしよう。
- 上記に関連して、ローカルに保存を終えたもの、GCSにアップロード済などが記録されるようにしておきましょう。
- 大量データの移行作業は完全性の確認が難しいです。
- 並列処理をうまく使おう。ただし、並列処理一般に言えることですが、ワーカー間で競合しないよう注意する必要があります。
- 例えば、同じファイルに複数のワーカーで書き込みにいってしまったり、別のワーカーが書き込み途中なのに、アップロードしてしまったりといったことが考えられます。
結果
今回のケースの場合、我々はc5d.largeインスタンスを1台だけ用意し、時間にして約20日間、コストにしておよそ1.5万円（インスタンス料金やデータ転送量含め）で実現することができました。
我々は、期限に余裕があったので、およそ20日間かけましたが、EC2インスタンスのスケールアップやスケールアウトでいくらでも短くできます。
まとめ
- 大量データをBQに格納する際は、まずはGCSにファイルを置き、ロードするようにしましょう。
- GCSのファイルパスは、テーブル名とパーティション日付でディレクトリ分けしよう。
- どれくらいのファイルPUTが発生するかは事前に見積もり、ファイルPUTが大量に発生しないようなプログラム設計をしましょう。
- AWSのEC2インスタンスで動かす場合、ローカルに大量の書き込みができるインスタンスストア対応インスタンス
c5d系を使おう。
様々な場所においてあるデータを一部加工し、BQにロードするという需要はそこそこあると思いますのでぜひとも参考にしてください。