Help us understand the problem. What is going on with this article?

AWS認定Big Data勉強記 - 5: Amazon EMR

More than 1 year has passed since last update.

みなさん、こんにちは、えいりんぐーです。

今回はEMRについてまとめます。

Q: Amazon EMR とは何ですか?

Amazon EMR は、企業、研究者、データアナリスト、および開発者が、簡単に、そして費用対効果の高い方法で、莫大な量のデータを処理することができるようにするウェブサービスです。Amazon Elastic Compute Cloud (Amazon EC2)、および Amazon Simple Storage Service (Amazon S3) のウェブスケールインストラクチャで実行されるホストされた Hadoop フレームワークを使用します。

資料集

基本的に以下の資料を参考にしています。

仕組み

Hadoopフレームワーク、マスターが多数のコアを管理して分散処理する。

ノード セキュリティグループ 役割
Master Node Master Instance Group - コアノードやタスクノードの監視。ネームノードやリソースマネジャー (YARN) が動く。Failover非対応。sshのポートが開いている。
Core Node Core Instance Group HDFSをアタッチされたボリュームとして持つ。計算を実行する。データノードが動く。追加したり、HDFSやCPUやRAMを増設できる。基本的にマスターとしか通信しない。
Task Node Task Instance Group HDFSを持たないコアノード。増設がより自由。スポット価格やインスタンスタイプを調整できる。基本的にマスターとしか通信しない。
YARN: Yet Another Resource Negotiator
  • Resource Manager
    • マスターサーバーで動く
    • スレーブのリソースを管理
    • ジョブ管理はしない
  • Node Manager
    • スレーブサーバーで動く
    • サーバーのリソースをRMに報告
    • サーバー上のコンテナの管理
  • Container
    • スレーブのリソースが切り出されたもの
    • NMによって起動
  • Application Master
    • ジョブ全体を管理するコンテナ

EMRにおいては、MapReduceやHiveやSparkへ計算資源を割り振ってくれる。共通のリソースを分割して扱ってくれる。

特徴と機能

  • EMRFS
    • S3をHDFSのように扱える。
      • CPUとストレージの分離
      • S3の暗号化も対応
    • Consistent View
      • 新しいオブジェクトのプット → Read-after-write consistency
      • 上書きプット/デリート → 結果整合性
      • DynamoDBにメタデータを格納
  • Bootstrap Action
    • ノード起動時に実行されるスクリプト
      • Bash, Ruby, Python, etc
      • S3におく
      • 引数も自由
  • Step
    • 任意のタイミングで追加・設定できる処理
      • HQLとか
    • S3のURIやローカルファイルを指定して実行
      • jar, Streaming, Hive, Pig, Spark, bash
  • ジョブ実行
    • クラスター外
      • Step API, Lambda, Data Pipeline, Airflowなどから
    • クラスター内
      • マスターノードにssh
      • AWS System Managerから
      • Hive, Spark, Zeppelin, Hue, Livy, Oozie
  • Private Subnetの利用
    • S3/DynamoにはVPCエンドポイントが必要
    • それ以外にはNAT GW
    • Security Groupが必要
  • 暗号化
    • at-rest: サーバーサイドやクライアントサイドで暗号化
    • in-transit: TLS
    • セキュリティ設定やCloudFormationで
  • タグ
    • タグをIAMポリシーのコンディションに指定してアクセス管理できる
  • オートスケール
    • yarn.resourcemanager.decommissioning.timeout
      • defalut: 1 hour
  • HBaseのデータストアとしてのS3
    • 可用性・耐久性が高くなる
  • Cloud Watch Event
    • EMRのイベントをモニタリングできる
  • インスタンスフリート
    • スポットインスタンスをまとめて管理
  • カスタムAMIを利用したクラスター起動
    • 追加ソフトウェアを事前にロードできるので設定が楽
  • アプリケーション
    • サポート
      • クラスタ作成時にオプションをつけてインストール可能なもの
    • カスタム
      • 事前にApache Bigtopに追加してBootstrapでデプロイ
      • カスタムAMIを使う

ポイント

  • Hive Metastoreは重要
    • SparkやPrestoからも参照可能
    • MetastoreのMySQLをクラスタ外にも作成可能
    • Glue Data Catalogを利用可能
  • 列指向ファイルフォーマット
    • ORCとParquet
      • I/O効率と圧縮効率がいい
  • HiveでKinesisのデータ処理
    • Kinssisが保持するデータをHiveのテーブルとして扱える
      • アーカイブ用途など
    • StreamのShard毎にMapperがデータを読み出す
  • Spark
    • 分散インメモリ処理フレームワーク
    • アプリケーション群
      • SQL
        • HQL互換
      • Streaming
        • Discretized Streamと呼ばれる、高レベル抽象表現
        • マイクロバッチ処理
        • KinesisのストリームをEMRに流せる
          • KCLがバックエンド
          • 数秒〜数分、ニアリアルタイム
      • MLlib
      • GraphX
  • IAMロール: 2つ必要
    • EMRロール
      • EC2を起動するのに必要
    • EC2 instance profile
      • EC2がS3などにアクセスするのに必要
  • Kinesis統合
    • Kinesisコネクタというものがある
      • Kinesisストリームから直接データの読み取りとクエリを実行できる
      • いちいち独立したアプリケーションを開発する必要がなくなった
    • Cronなどでスケジュールされたジョブ実行ができる
    • MapReduce自体はバッチ処理なので、ストリームを複数のバッチに分割する
      • 各バッチを反復計算と呼ぶ
      • バッチの失敗に際して再試行するので、冪等性を保証する
    • 1つの反復計算で複数のクエリを実行できる
      • kinesis.checkpoint.iteration.no を設定する
      • 論理名を別にすることで、並列クエリ実行ができる
    • 連続ストリーム処理はできない
      • Twitter StormやSpark Streamingを使えばなんとか
    • 複数のKinesisストリームを結合することもできる
    • ストリームにデータがないとき
      • kinesis.nodata.timeout の時間分だけレコードの取得を試行する(ポーリング)
      • 時間が経ったらポーリングを停止して、すでにあるレコードだけ処理する
      • 新しくレコードが来たら kinesis.iteration.timeout 分だけ待ってポーリングする

ベストプラクティス

データ移行
  • s3distcp を使う
    • distcp はHadoopエコシステムでよく使われるコピーコマンド
    • s3distcp はS3に最適化されている
  • AWS Import/Exportを使う
    • データをストレージに入れて物理的に移行する
    • データが大きい時に有効
  • AWS Direct Connect
    • AWSとオンプレミスを専用回線でつなぐ
    • パブリック環境とプライベート環境とでネットワークを分離できる
データ集約
  • 多数の小さいデータではなく、少数の大きいデータにまとめる
  • S3のオブジェクトは、チャンクサイズに分割されてEMRに読み込まれる
    • だいたいチャンクサイズは64MBとか
    • データの圧縮形式が分割に対応していると良い
      • LZOやBZIP2
  • 集約されたデータのサイズは均等だと良い
  • マッパーとレデューサーの間の中間データも圧縮できる
    • データ量は REDUCE_SHUFFLE で確認できる
    • 圧縮は mapreduce.output.compress を true にする
  • マッパーが書き出すファイルも圧縮できる
    • mapred.compress.map.output を true にする
  • データのパーティション
    • 大きすぎるデータは逆に非効率なので、程よいサイズに分割する
    • 1時間単位で分析するようなデータなら、日時でデータを分割する
インスタンスタイプ

EBSをアタッチできるようになったので、M4やC4ファミリーのインスタンスを使えるようになった
ただし暗号化されたEBSをサポートしていない
要件として、計算が重いのか、メモリが必要なのか、I/Oが重要なのかに応じてインスタンスを選ぶ

EC2 A1, T3, T2, M5, M5a, M4 C5, C5n, C4, z1d R5, R5a, R4, X1e, X1, ハイメモリ P3, P2, G3, F1 H1, I3, D2
A1: ARMアーキテクチャ。Tx: バースト可能、処理のスパイクに対応。Mx: 汎用 コンピューティング最適化。CPUが強い メモリ最適化。メモリが強い。 GPU ストレージ最適化。ローカルストレージが強い。データのスループット良い
EBS gp2 io1 st1 sc1
汎用 プロビジョンドIOPS。低レイテンシー・高スループット スループット最適化HDD。HDFS向き。 Cold HDD。低アクセス。HDFS向き
  • 一時的なクラスター: Transient Cluster
    • 処理が終了したらシャットダウンするもの
      • HDFSがプライマリストレージでない
      • 試行錯誤する作業
  • 永続的なクラスター: Permanent Cluster
    • 処理が終了しても起動状態にしておくもの
      • ジョブが頻繁で定期的
      • 相互のジョブに依存関係がある
  • オンデマンドインスタンス
    • 使用する処理能力に応じて課金
    • 一時的なクラスターに向いている
  • リザーブドインスタンス
    • 予約するインスタンスに応じて予約金を払う
    • オンデマンドより単価は下がる
    • 永続的なクラスターに向いている
  • スポットインスタンス
    • 非稼動のEC2インスタンスに入札して利用する
    • オンデマンドより単価が下がる
    • 一時的なクラスターやタスクノードの追加に向いている
アーキテクチャ
  • S3をプライマリストレージにする
    • メリット
      • データの耐久性
      • コンピューティングとストレージの分離
      • データセキュリティやライフサイクルの管理
      • ストレージのスケーリング
      • Transient Clusterの利用
      • 複数クラスターでのデータ共有
    • デメリット
      • 試行錯誤する作業の場合は、S3とのネットワークがボトルネックになる
  • S3とHDFSの併用
    • S3に保存しつつ、実行時にHDFSにコピーする
    • メリット
      • S3の耐久性と可用性
      • Transient Clusterの利用
      • 反復的な作業の効率化
    • デメリット
      • データコピーによる処理時間増加
  • HDFSがプライマリ、S3をバックアップ
    • メリット
      • S3からデータをコピーする必要がない
      • 頻繁な処理の高速化
    • デメリット
      • データの耐久性
  • 伸縮自在なクラスター
    • マスターノードと必要最低限のコアノードを設定する
    • タスクノードをデータ処理の要求に応じて追加・削除する
    • インスタンスフリート

セキュリティ

  • IAMによるリソースのアクセスコントロール
  • セキュリティグループによるノードに対するネットワーク
  • S3-SSEやKMS、CSE-KMSを利用したat-rest暗号化
  • TLSを利用したin-transit暗号化
  • オープンソースアプリケーションやLUKSを利用したローカルディスクでの暗号化
  • Kerberos認証

細かいこと

  • TCPウィンドウスケーリング
    • TCP ウィンドウスケーリングを使用すると、64 KB を超えるウィンドウサイズをサポートすることにより、オペレーティングシステムおよびアプリケーションレイヤーと Amazon S3 との間でネットワークスループットのパフォーマンスを向上させることができます。TCP セッションの開始時に、クライアントは、そのサポートされているウィンドウ WSCALE 係数をアドバタイズし、Amazon S3 は、アップストリーム方向でサポートされている受信ウィンドウ WSCALE 係数を返します。
  • TCP選択的伝送確認
    • 一部は正しく受信しましたよ、と送信側に信号を送る。伝送効率が上がる。
  • 高度な最適化
    • データ構造の設計
      • 適切にパーティショニングする
    • Hadoopは基本的にバッチ処理フレームワークなので、時間制約がある場合はStormやSparkを使う
    • クラスターの細かいチューニングよりノードの追加が簡単で有効
    • マッパーの改善
      • マップ数の削減
        • 必然的に大きなデータを処理することになるので、マッパーあたりの処理量が向上する
      • マッパー出力の圧縮
      • マッパーのディスク書き出しの回避
    • レデューサーの改善
      • アイドル状態レデューサーの削減
      • レデューサーメモリの増加
brainpad
ブレインパッドは、2004年の創業以来、データによるビジネス創造と経営改善に向き合ってきたデータ活用・分析企業です。
http://www.brainpad.co.jp/
Why not register and get more from Qiita?
  1. We will deliver articles that match you
    By following users and tags, you can catch up information on technical fields that you are interested in as a whole
  2. you can read useful information later efficiently
    By "stocking" the articles you like, you can search right away