(レポート) BDT305: Amazon EMR Deep Dive & Best Practices #reinvent

本セッションでは、最新のEMRリリース4.1の高度な機能に関するベストプラクティスとチューニングについて解説されています。EMRでHiveを利用の方はぜひ、このセッションは抑えておいてください。

BDT305: Amazon EMR Deep Dive & Best Practices

このセッション

• 最新のEMRリリースの更新
• EMRの高度な機能に関する情報
• EMRのコストを低下させるためのTips
• インフラはAmazon EMRとAmazon S3をマルチペタバイトのデータウェアハウスとしてどのように使っているかの徹底解説

Amazon EMR

• Hadoop, Spark, Prestoのクラスタや、Apache/Hadoop スタックのその他アプリケーショションの管理
• Amazon S3, Amazon DynamoDB, Amazon Kinesis, Amazon Redshift, AWS KMSのコネクターである、EMRFSを経由してAWSプラットフォームに統合される
• AWS IAM roles, KMS, S3 client-side encryption, Hadoop transparent encryption, Amazon VPC, and HIPAA-eligibleのサポートでセキュアです。
• クラスタのサイズを変更をサポート済みで、低コストを支援するためのAmazon EC2スポット・マーケットと統合している

New Features

EMR Release 4.1

• • Hadoop KMS の透過的HDFS暗号化のサポート
• • Spark 1.5, Zeppelin 0.6
• • Presto 0.119, Airpal
• • Hive, Oozie, Hue 3.7.1
• • 起動と設定の Simple APIs

Intelligent Resize

• 利用可能な容量キャパシティに基づいて増分スケールアップ
• スケールダウン変更前に完了する作業を待ち
• コアノードとHDFS及びタスクノードを拡張することができる

Amazon S3上のEMR File System (EMRFS)の活用

永続データストアとしてAmazon S3

• コンピューティングとストレージの分離
• データ損失無しでEMRクラスタのリサイズとシャットダウン
• Amazon S3の内の同じデータでのを複数のEMRクラスタから利用する
• 著しい技術の進化に伴い、分析インフラストラクチャを進化させます

EMRFS はS3の扱いが容易になる

• Read-after-writeによる一貫性
• リスト関連操作が高速
• エラーハンドリングオプション
• Amazon S3 暗号化のサポート
• アプリケーションから透過的：s3://

HDFS から Amazon S3 へ

CREATE EXTERNAL TABLE serde_regex(
host STRING,
referer STRING,
agent STRING)
ROW FORMAT SERDE 'org.apache.hadoop.hive.contrib.serde2.RegexSerDe'
);

• HDFS LOCATION: ‘samples/pig-apache/input/'
• S3 LOCATION: 's3://elasticmapreduce.samples/pigapache/input/'

オプションのEMRFSメタデータレイヤを使用による一貫性のあるビューと高速リスト

EMRFS クライアントサイド暗号化

クライアント暗号化の暗号化キーの作成は S3のユーザ提供キーによるクライアントサイド暗号化 (CSE) を使い倒す を御覧ください。

必要であればHDFSはこれまで通り使えます

• 反復ワークロード
  • 複数回同じデータセットを処理する場合
  • Spark と RDDsの使用を検討してください
• ディスクI / Oが集中するワークロード
  • Amazon S3の上のデータを永続化し、処理のためにS3DistCpを使ってHDFSへのFrom/To コピーする

ストレージのための最適化

ファイルフォーマット

行指向

• テキストファイル
• シーケンスファイル
• 書き込み可能オブジェクト
• Avroデータファイル
• スキーマによる記述

列指向

• オブジェクレコードカラム(ORC:Object Record Columnar)
• Parquet(Hadoop用カラムナストレージ)

考慮すべき要素

• 処理およびクエリツール
  • Hive, Impala and Presto
• スキーマの進化
  • スキーマ用のAvroとストレージ用のPresto
• ファイル形式「分割性」
  • JSON / XMLファイルを避ける
• レコードとして使用する

ファイルサイズ

• 小さなファイルを避ける
• 100メガバイトより小さいもの
• 各Mapperは、単一のJVM
• CPU時間は、 JVM/Mapperを起動するために必要です
• より少ないファイルで、ブロックサイズを一致させる
• S3は呼び出しを少なくする

小さいファイルの取扱

HDFSのブロックサイズを小さく、e.g. 1MB (default is 128MB)

• --bootstrap-action s3://elasticmapreduce/bootstrapactions/configure-hadoop --args “-m,dfs.block.size=1048576”

Better：S3DistCpを使うと小さなファイルを結合する

• S3DistCpは、より大きなものに小さく、入力ファイルを結合するためのパターンとターゲットパスを取る
• ターゲットサイズと圧縮コーデックを指定する

圧縮

常にS3の上のデータファイルを圧縮

• S3-EMR間のネットワーク・トラフィックを削減
• あなたの仕事をスピードアップ

MapperとReducer出力を圧縮

EMRは、Hadoop1はLZO、Hadoop2はSnappyによって、ノード間のトラフィックを圧縮

適切な圧縮を選択

• 時間に敏感な、より高速な圧縮は、より良い選択であります
• 大量のデータ、使用スペース効率的な圧縮
• 複合ワークロード、GZIPを利用する

Amazon EMR のコスト圧縮 Tips

• 永続的なデータストアとしてS3を使用する - クエリに Presto, Hive, Spark等を使用する
• 計算が必要なときにのみ支払います（不必要なときはクラスタを起動しない）
• 80％以上節約するには、Amazon EC2のスポットインスタンスを使用する
• 安定したワークロードのためには、Amazon EC2リザーブドインスタンスを使用する

FINRA の事例

米国で最大の独立系の証券規制当局の一つである金融取引業規制機構「FINRA」は、毎日行われているの金融取引慣行-十億を見て、規制に議会によって課されています。市場のダイナミクスは、FINRAが、現在のままにするための課題となって急速に変化します。そこで同社は分析し、毎日約300億市場のイベントを格納するためにAmazon Webサービス（AWS）へのプラットフォームを移動しました。

最大750億イベント/日、５PB超えのストレージ

データ

• S3はレコードの永続性のあるシステム
• コンシステントビューのEMRFSを使う
• パフォーマンスのためデータを分割する
• 大きなファイルは256MB以上が効率が良い
• コンパクトな小さなファイルは100MB
• FINRAのデータマネージャーはストレージとコンピュートクラスタ間のデータをオーケストレートする

ファイルフォーマットと圧縮

• アーカイブは、テキスト形式でS３とGlacierに保存
• ベストフィットのための圧縮アルゴリズムの選択
• パフォーマンスのための行選択、もしくはは列指向フォーマット
• 列指向の成果は、Predicate pushdown機能、不要なカラムの排除、複数のクエリエンジン（Hive, Presto, Spark）の提供

パーティションとクエリの戦略

Hiveテーブル作成の例

レコードごとにハッシュパーティションナンバーを算出： ((pmod(hash(symbol), 100) * 10) + pmod(firm, 10))

create external table if not exists NEW_ORDERS (…)
partitioned by (EVENT_DT DATE, HASH_PRTN_NB SMALLINT)
stored as orc
location 's3://reinvent/new_orders/'
tblproperties ("orc.compress"="SNAPPY");
alter table NEW_ORDERS add if not exists
partition (event_dt='2015-10-08', hash_prtn_nb=0) location
's3://reinvent/new_orders/event_dt=2015-10-08/hash_prtn_nb=0/’
…
partition (event_dt='2015-10-08', hash_prtn_nb=1000) location
's3://reinvent/new_orders/event_dt=2015-10-08/hash_prtn_nb=1000/’
;

EMR/S3で構築したHive

競合製品の比較、"EMR vs Greeplum"の結果、EMRが若干劣る結果に...

パーティションニングは良いが、しかし注意が...

select … from NEW_ORDERS where
EVENT_DT between '2015-10-06' and '2015-10-09’
and FIRM = 12345
and (pmod(HASH_PRTN_NB, 10) = pmod(12345, 10)
or HASH_PRTN_NB = 1000) -- 1000 is always read

hash_prtn_nbのまわりでpmodを使うことは、Hiveが剪定のために返される何百万もの結果metastoreでターゲットとされたクエリを使うのを妨げている。

enumerationによるクエリの最適化

select … from NEW_ORDERS where
EVENT_DT >= '2015-10-06' and EVENT_DT <= '2015-10-09’
and FIRM = 12345
and (HASH_PRTN_NB = 5 or HASH_PRTN_NB = 15
… or HASH_PRTN_NB = 985 or HASH_PRTN_NB = 995
or HASH_PRTN_NB = 1000) -- 1000 is always read

剪定問題を回避するには不十分なIN句を使用した。 明示的にすべてのパーティション大幅に改善クエリ計画時間を列挙する

データセキュリティ

• Required to have encryption of all data both at-rest, and in-transi
• 停止中、移動中のすべてのデータの暗号化することが要求された
• S3サーバサイドの暗号化を評価し、目的のために適切であると決定された
  • LUKS（ディスク暗号化）、Mapper／Reducerのテンポラリの暗号化、サーバーやデータがロスしてもS3がある
• 唯一のクライアントアプリケーションがマスターノードに接続を保証するためにセキュリティグループを使用する
• Hive認証/認可は、使用シナリオに必要ではなかった
• HDFSの透過的な暗号化（ Hadoopのを2.6+ ）評価

適合試験の選択

• HDFSは、ユースケースでは法外な費用がかかりました
• データの局所性は、この規模には実用的で望ましくありません
• 階層型ストレージの検討の余地あり

Darwin rules: Adaptation

新しいインスタンスタイプを活用してください ワークロードに適切なインスタンスタイプ（複数可）を見つける 大きなノードの小さいクラスターを好む：e.g. 4XL パーティションの数百万が、より多くのメモリがマスターノードのために必要とされる（HS2） コンソールよりもCLIベースのスクリプトを使用する → Infrastructure as code

Beat the incumbent

競合製品の比較、"EMR vs Greeplum"の結果、EMRが36%の改善、並列性が優位、低コストという結果に改善した。

適切なクラスタのサイズ

• 一時的に起動するユースケース： ETL 、バッチ分析
• 常に起動するユースケース：インタラクティブ分析
• ボトルネックを回避するために、 5 ： 1のタスクへのコアの比率を維持
• より高い安定性を確保するために、オンデマンドの価格の上にスポットを入札考えてみましょう

一つのメタストアは、それらすべてを規定する

• 共有Hiveメタストアサービスを作成することを検討する
• マルチAZ RDSによるフォールトトレラントとDR
• 表およびパーティションのメタストアの負担を軽減する
• 分割されたmetastoresはHive 0.13.1 、Hive 1.0とPrestoのために必要とされています
• メタストアの更新を調整するためにFINRAデータ管理サービスの活用する

監視、習得、最適化

• ワークロード管理の活用：フェアなスケジューラ
• コードは、ボトルネックを解消するために必要に応じてリファクタリング
• Set hive.mapred.reduce.tasks.speculative.execution = FALSE ：Mapper Reducerを経てS3の外部テーブルに書き込んだ時
• 小さなテーブルを結合するとき、ブロードキャストジョインを使用する(SET hive.auto.convert.join=true)
• EMRのステップAPIは前方のシンプルなジョブキューイングを動作し、より複雑なジョブのためOozie（ワークフロースケジューラ）を使用する

インパクト

• 削除された障害
  • 「この大きさのデータ分析の前に技術チームからの介入を必要とする。 」
• 好奇心のコストを下げた
  • アナリストは、すぐに時間をかけて順番に何が起こるかの全体像を取得することができる
  • 規則違反が発生したか否かの意思決定を通知するのに役立つ。
• 拡張性は、私たちが数ヶ月とは対照的に、日中のデータの年を処理し、スポット市場を利用してお金を
• 節約することができます
• 別に妥協することなく、バッチおよび対話型のワークロードを最適化する
• 増大したチームの配信速度

Recap

• レコードの耐久性のあるシステムとしてはAmazon S3を使用する
• できるだけ多くの一時的なクラスタを使用する
• クラスタのサイズを変更し、より効率的に、容量、パフォーマンス、およびコストを管理するためにスポットを使用する
• 新しいインスタンスファミリに移動し、パフォーマンスの利点を活用する
• サイズ変更またはインスタンスタイプを変更するかを決定するために監視する
• 複数のEMRクラスター間で、RDS内の永続Hiveメタストアを共有する
• 将来的にはクエリ・エンジンまたは実行フレームワークを切り替えできるように準備する
• 予算時間は前には不可能だった規模での新しいツール＆エンジンのための実験する

最後に

EMRのHiveに関するベストプラクティスと導入事例を通じたチューニングについて解説されたセッションでした。将来的にはクエリ・エンジン（Hive, Presto, Spark）または実行フレームワークを切り替えできるように準備することで、他のエンジンにおいても今回のプラクティスを活かしたいところです。