マイクロアドでETL処理周りなどデータエンジニアリングを担当している高橋です。
マイクロアドではアドホックな分析用途にImpalaを導入していますが、集計元テーブルの準備、具体的にはメタデータ周りの設定に手こずった経験があります。
今回はどこに苦しんだのか、どう解決したのかなどをお話ししたいと思います。
前提
マイクロアドでは広告配信ログを共通のHadoopクラスタ(以後、CDHを利用しているのでCDHクラスタと称します)で処理しています。詳細は以下のブログ記事をご参照ください。
共通のCDHクラスタではHiveクエリエンジンを使用してバッチ処理を行っています。
Hiveはバッチ用途で使用する分には最適なのですが、MapReduce処理特有の分散処理がDiskに書き込まれて処理時間がかかるため、分析用途には向いていません。そのため別途データ分析用にCDHクラスタを用意しています。こちらではImpalaクエリエンジンをメインに使用して分析処理を行っています。
導入の経緯については以下のブログ記事をご参照ください。
処理の概要
データ分析用のCDHクラスタで使用するテーブルは、共通のCDHクラスタから転送する必要があります。図にすると以下のようになります。
メタデータ更新問題
テーブルの転送フロー修正前
転送バッチ開発開始当初、テーブルごとの転送フローは以下を想定していました。
- Hiveクエリでパーティションを追加する
ALTER TABLE {転送対象テーブル} ADD IF NOT EXISTS PARTITION ({パーティションカラム} = {パーティションの値});
- distcpコマンドを用いてCDH間ファイルコピーを行う
hadoop distcp -m {mapper数} -bandwidth {バンド幅} -pb -overwrite -delete {共通CDHクラスタ上転送元パス} {分析用CDHクラスタ上転送先パス}
既存のCDH間転送に則ったフローのつもりだったのですが、これで実際に転送を行うと問題が起こりました。
Hiveでは転送元と転送先の件数が一致するのに対し、Impalaでは転送先の件数が0件になってしまうのです。
コピーされた同じファイルを参照しているはずなのに、いったい何が起こっているのでしょうか。
メタデータと統計情報の紐づけ
HiveやImpalaなどのクエリエンジンにはメタデータや統計情報という概念が存在します。
メタデータはテーブル構造やHDFS上のデータのパス等の情報、統計情報はテーブルのレコード数やファイルサイズ等の情報です。どちらもクエリ実行時に欠かせない情報となっています。
今回の転送フローを振り返ってみると、
- Hiveクエリでパーティションを追加する
- Hiveのメタデータ更新を行う
- distcpコマンドを用いてCDH間ファイルコピーを行う
- 単なるCDH間のファイルコピー
となり、Impala側から見ると全く検知できていないことになります。
Impalaがレコード数を確認するために統計情報を探しに行っても見つからない状況になっているのです。
一方でHive側も統計情報が更新されないため、件数が確認できたとしても効率の悪いクエリ実行になってしまいます。
テーブルの転送フロー修正後
以上を踏まえて修正した転送のフローは以下の通りです。
- Hiveクエリでパーティションを追加する
ALTER TABLE {転送対象テーブル} ADD IF NOT EXISTS PARTITION ({パーティションカラム} = {パーティションの値});
- distcpコマンドを用いてCDH間ファイルコピーを行う
hadoop distcp -m {mapper数} -bandwidth {バンド幅} -pb -overwrite -delete {共通CDHクラスタ上転送元パス} {分析用CDHクラスタ上転送先パス}
- Hiveの統計情報を更新
ANALYZE TABLE {転送対象テーブル} PARTITION ({パーティションカラム} = {パーティションの値});
- Impalaのメタデータを更新
ALTER TABLE {転送対象テーブル} RECOVER PARTITIONS;
- Impalaの統計情報を更新
COMPUTE INCREMENTAL STATS {転送対象テーブル};
修正前はHiveのメタデータのみを更新していましたが、修正後はHiveの統計情報およびImpalaのメタデータ、統計情報を更新するようになりました。1
既存の転送フローを改善し、効率的なクエリ実行ができるような準備が整ったと思います。2
メタデータ大きすぎる問題
突然のERROR
テーブルの転送を開始してからしばらくたつと、 COMPUTE INCREMENTAL STATS {転送対象テーブル}; を実行する際に以下のようなERRORが発生しました。
Query: compute INCREMENTAL STATS {転送対象テーブル}
ERROR: AnalysisException: Incremental stats size estimate exceeds 200.00MB. Please try COMPUTE STATS instead.
Could not execute command: compute INCREMENTAL STATS {転送対象テーブル}
文面は分かるのですが、どうしてこんなことが起こるのか最初は理解ができませんでした。
ERROR文に記載のあるCOMPUTE STATS を使用するとクラスタに負荷をかけるため、何とか避けたいところです。
大きくなりすぎたオンメモリメタデータ
Impalaでは、各Worker Node上のImpalaデーモンが以下のメタデータをメモリにキャッシュしています。
- テーブル数
- パーティション数
- ファイル数
- ブロック数
- 各テーブルのカラム数 × パーティション数
今回の件に関連した部分だと、各テーブルのカラム数とパーティション数が特に重要です。各テーブルのカラム数とパーティション数は掛け算されてキャッシュに影響を及ぼします。
ERRORになったテーブルは構造化されていない非常に大きなテーブルでカラム数も膨大でした。
1パーティション1カラムあたり約400バイトのメタデータをキャッシュする必要があるので、統計情報更新に必要なメモリ上限を超えてしまったのです。
参考リンク: Apache Impalaパフォーマンスチューニング #dbts2018 101ページ
解決方法
転送対象テーブルの変更可能な部分を検討したところ、パーティションカラムの削減に行きつきました。
今まで複数カラムを組み合わせてパーティションを構成していたのですが、一部をパーティションから削除し通常のカラムとして使用する形に変更しました。
パーティションカラムが削減されたことにより無事メタデータが削減され、既存の統計情報更新クエリが流せるようになりました。3
なおImpalaデーモン実行時のパラメータである --inc_stats_size_limit_bytes でキャッシュサイズ変更を行う方針もありますが、上げすぎるとヒープ使用量が急増したりImpalaがクラッシュする恐れがあるので、あまり得策ではありません。
参考リンク: Maximum Serialized Stats Size - Table and Column Statistics | 6.3.x | Cloudera Documentation
まとめ
ImpalaはHiveよりもメタデータや統計情報に対して敏感で、きちんと設定してやらないと動作に影響が出るようです。 今後もインフラチームと協力しつつ、更に使いやすい分析基盤を構築していきたいです。
参考リンク
Apache Impala - Interactive SQL | 6.3.x | Cloudera Documentation
Apache Impala Guide (impala-3.2)
Apache Impalaパフォーマンスチューニング #dbts2018
-
一番最初に新しいテーブルを追加した際には、
INVALIDATE METADATA {転送対象テーブル};といったクエリを流す必要があります。 ↩ -
後から確認されたことですが、Impala側でメタデータや統計情報を更新したらHiveにも伝搬するようです。Impalaのメタデータ、統計情報は更新クエリ実行時にHiveメタストアのメタデータ、統計情報を更新して同期しているためです。Hiveから更新クエリを実行してHiveメタストアのメタデータ、統計情報を更新してもImpalaには同期されません。↩
-
パーティションカラムの設定はクエリ実行時のパフォーマンスに直結するため、安易な削減には注意が必要です。↩
