※ この記事はお客様に寄稿いただき AWS が加筆・修正したものとなっています。

本稿は株式会社 NTT ドコモ モバイル空間統計 における AWS Glue を活用したリアルタイムストリーミング処理の取り組みについてご紹介します。取り組みのご紹介は全二回となっており、第二回となる本編では Glue を新規採用する際の開発面での課題と Glue ストリーミングジョブのパフォーマンス改善の取り組みについてご紹介します。

第一回はコスト削減の取り組みについてご紹介していますので、こちらも併せてご覧ください。

以下、本システムにおける構成図を再掲します。

Glue のパフォーマンス課題

Glue ジョブパフォーマンス改善のアプローチ

モバイル空間統計では大量に流れ込む位置情報データを遅延なくリアルタイムに処理し続けられることが求められます。今回 Glue へ移行するにあたり、Spark アプリケーションとして実装し直して既存と変わらないパフォーマンスを出せるかという点が最も重要な課題となっていました。

パフォーマンスの改善には、まずプロファイルを取ることが重要です。Python にはいくつかのプロファイラが存在しており、それらを利用してプロファイルを取得し、パフォーマンス改善やデバッグを進めることができますが、 Spark では同じ方法で取得することは出来ません。今回私たちが採用した手段は以下の 2 点です。

1. Spark History Server(Spark UI)による Glue ジョブ実行状況、ノード稼働状況の可視化

Glue ジョブの詳細や各ノードの状況を把握するための方法としては Spark History Server(Spark UI) ^※1を利用できます。Job details で実行時にログを出力する設定を行えば、ジョブ実行後に指定の S3 にログが出力されます。ブラウザ上で Spark UI にこのログを読み込ませ、ジョブ実行内容を可視化します。この Spark UI は Docker イメージが公開されていますので、ローカルマシン上で簡単にコンテナ実行することで利用可能です。また、AWS マネジメントコンソールからもワンクリックで起動することもできます。主に以下の目的でパフォーマンス改善に活用しました。

• 使用効率の観点から、各ワーカーが均等に稼働できているか(executor ごとに処理対象のデータ量に偏りはないか)を把握する。
• Spark における各ジョブ(Glue ジョブではないことに注意)の処理時間を確認し、どの処理がボトルネックとなっているかを特定する。

（※1）Spark UI を利用することで、各ワーカーノードでのタスクの分散状況を可視化することができます。これにより、効率的な分散処理が行われているかどうかを確認し、行われていない場合はワーカーノードの数を調整できます。

以下の図は Spark UI のスクリーンショットで、ワーカーノードのCPU上で動作したあるタスクの実行時間を示す棒グラフです。この例では、 2vCPU のワーカーノードで動作させた際に分散処理がうまく働いていないことがわかります。タスクが終了したのちに新たなタスクが始まっている様子が図の赤枠の部分で確認できます (緑色の線が一度途切れた後、新たな線=タスクが始まっています)。これは、処理が並列で行われていないことを示しています。この現象は、タスク数(入力データの分割数)に対して Glue 側の vCPU コア数が足りていないことが原因です。このような状況を改善するために、ワーカーノードの台数やワーカータイプの変更を検討します。

以下の図は改善結果です。ワーカータイプを 2vCPU(G.025X) から 4vCPU (G.1X)に変更し、最大ワーカー数もタスク数に合わせて増やしました。その結果、上図では一部のCPUが複数回タスクを処理する非効率な状態でしたが、下図では全てのCPUがそれぞれタスクを1回で並列に処理する効率的な分散処理の状態に改善できました。

2. Glue ジョブメトリクスによるジョブのパフォーマンス計測

Glue の Job details にはジョブメトリクスを集計する設定があり、これを有効にすると、メトリクスが集計されます。今回ストリーミングジョブのパフォーマンスを評価するため、特に把握したかったのはマイクロバッチ処理ごとの処理時間です。ウィンドウサイズ(マイクロバッチが 1 回あたりで処理するストリーミングデータの時間枠)に対して、それよりも短い時間(理想は 70% 未満の時間)でバッチ処理が完了していれば、リアルタイムに対して遅延なく処理ができていると判断できます。Glue ジョブメトリクスの中に batchProcessingTimeInMs があり、マイクロバッチごとの処理時間(ms)が分かります。これをもとにして、ウィンドウサイズに対するマイクロバッチの処理時間の比率を示すメトリクス(Process rate という名前をつけています)を算出し、これを評価することにしました。

Process Rate = batchProcessingTimeInMs * 1000 / windowSize(s)

この Process Rate が 1.0 を超えない状態で安定稼働するかどうかを判断基準としてパフォーマンス改善を進めました。

Spark アプリケーションのパフォーマンスの改善

Spark アプリケーションとしてのパフォーマンスチューニング

PySpark というライブラリを用いれば、 Python で Spark アプリケーションを実装できます。ただし、Pythonと同じ感覚で実装するだけではパフォーマンスが向上せず、Spark の特性(遅延評価の仕組み、変換とアクション、シャッフルとステージなど)を理解することが重要です。
Glue のパフォーマンスチューニングについては、 AWS が公開している Spark のチューニングに関するガイドなどに従って、内部設計の改善を進めました。このように、Spark UI でジョブの実行状況を可視化し、パフォーマンスを詳細に把握した上で試行錯誤を通じてパフォーマンス改善を進める必要がありました。

Kinesis Data Streams のシャード数に応じた Glue のワーカータイプ、ワーカー数の検討

今回 Glue ストリーミングジョブが接続する入力元は Amazon Kinesis Data Streams です。Glue のワーカータイプやワーカー数を設定する際には、Kinesis Data Streams のシャード数を考慮する必要があります。Spark アプリケーションとして効率的な分散処理をするため、全ての Executor ノードにデータが均等に行き渡る状態が求められます。全てのCPUコアを余すことなく利用するためには少なくとも Executor ノードの数(=Glue ワーカーごとの vCPU コア数の総数)以上にデータが分割された状態である必要があります。このデータの分割数(以下パーティション数)は、Kinesis Data Streams のシャード数と一致します。一方でパフォーマンス効率の観点では全てのパーティションを１並列で処理するために Glue 側では、Executor ノードの合計 vCPU コアが Kinesis Data Streams のシャード数以上となるように準備する必要があります。これによって必要なワーカー数が決定されますが、どのワーカータイプを選定するのが適切かという検討も必要です。ワーカータイプの選定に関しては、以下を考慮しました。

• DPU あたりの vCPU コア数(G.025X は他のインスタンスタイプと比べて、vCPU/Memory の比率が高く、コア数が多くメモリが少ない)
• ワーカータイプは Driver ノードと Executor ノードで別々に指定できないため、Driver ノードが過剰に高スペックとならないような考慮

これらの要素をもとに、最適なワーカータイプを決定しました。今回のワークロードではストリーミングデータ処理を行うため、個々のデータサイズが非常に小さく、メモリ量よりも CPU コア数の確保を優先すべきと判断し、G.025X のワーカータイプを候補として選定しました(G.025X は、ストリーミングジョブを想定したワーカータイプとして用意されています)。しかしながら、実際に G.025X で動作検証をしたところ、Driver 側で動く Python プロセスが使用するメモリが不足する問題が発生したため、最終的にはワーカータイプは G.1X を採用しました。もちろん G.2X でも動きますが、Driver のスペックとしてはオーバースペックとなり、ワーカー 1 台あたりの vCPU コア数も多くなり(G.2X の vCPU コア数は 8)、必要以上にコア数を確保して余らせてしまうことになる可能性が高く、コストが無駄にかかって非効率なため、このワーカータイプの選定は非常に重要です。^※2

(※2) Glue のワーカータイプは G.025X が2vCPU, 4GB メモリと 1:2 の比率になりますが、G.1X 以上のサイズのワーカータイプは 1:4 の比率となります。

まとめ

本記事では AWS Glue への移行に伴うパフォーマンス問題とその対応についてご紹介しました。

Glue 上で動作する Spark の特性を把握し、Spark UI によるジョブ実行状況を詳しく可視化することでパフォーマンスの改善を進めることができました。また、Kinesis Data Streams のシャード数、Spark がデータを読み込む際のデータ分割数との関係を考慮し、Glue のワーカータイプや最大ワーカー数を適切に決定できました。

コスト削減においては Glue と S3 の二つのサービスに対して効率的なコスト削減を実施しました。以上のような取り組みにより、リアルタイムで膨大なデータ処理を継続的に行うために必要なパフォーマンスをAWS Glueにて実現することができました。

今後の取り組み

今回のシステムではメンテナンス性・オートスケール機能の観点で AWS Glue を選び、クラウドへの移行・ビッグデータ処理の改善を進めることができました。また、 EMR on EC2 でスポットインスタンスを使い、コスト削減の検証を進めています。今後もメンテナンス性を維持しながら、さらなる運用コストの削減を目指していきます。

総括

第一回と本記事で「モバイル空間統計」における AWS Glue への移行を進めていく中での課題と解決策・ポイントについてご紹介しました。 AWS Glue により運用効率とスケーラビリティを両立したシステム運用が可能となりました。
本事例がAWS Glue を活用したリアルタイムストリーミング処理を検討している皆様への参考になれば幸いです。

著者

株式会社 NTT ドコモ コンシューマサービスカンパニー 第一プロダクトデザイン部 マーケティングイノベーション・カスタマーサクセス 土屋 慶和

株式会社NTTドコモでクライアントサーバシステムやspモード^®の大規模NWのアーキテクト検討やスーパー販促プログラム^®の立ち上げ・プロジェクトマネージャーを担当し、現在はモバイル空間統計・di-PiNK^® DMPのプロジェクトマネージャーとして従事しています。

NTT コムウェア株式会社 NTT IT 戦略事業本部 DigitalDesign&DevelopmentCenter DataManagementPartner部門 データビジネス担当 市川 大助

NTT コムウェアにて IT アーキテクトを中心にフルスタックエンジニアとしてシステムの開発、更改、運用に従事してきました。現在はマネージャーとして、ドコモグループの様々なデータ分析業務チームを統括しています。

NTT テクノクロス株式会社 IOWN デジタルツインプラットフォーム事業部 第二ビジネスユニット 岳野裕也

NTT テクノクロスにて、Python エンジニアとしてデータ分析システムの PoC・開発等に従事してきました。現在はモバイル空間統計関連システムの開発リーダー兼リードエンジニアとしてチーム管理や技術面の課題解決等を担当しています。

アマゾン ウェブ サービス ジャパン合同会社 平川大樹

ソリューション アーキテクトとして通信業界のお客様の AWS 活用 を支援しています。