Amazon EMRの入門チュートリアルを試してみた

# 背景・目的
[こちら](https://qiita.com/zumax/items/4b641f49616e03bd6c76)でEMRを整理しましたが、今回は、[ドキュメント](https://docs.aws.amazon.com/ja_jp/emr/latest/ManagementGuide/emr-gs.html)をもとにEMRのチュートリアルを試してみます。

# まとめ
- わずか数クリックでクラスタが起動できます。

# 概要
EMR を使用すると、ビッグデータフレームワークを使用してデータを処理および分析するクラスターをわずか数分でセットアップできる。
このチュートリアルでは、Spark を使用してサンプルクラスターを起動する方法と、S3 バケットに格納された簡単なPySpark スクリプトを実行する方法について試す。
次の 3 つの主要なワークフローカテゴリにおける EMR の必須タスクを取り上げる。
- 計画と設定
- 管理
- クリーンアップ

![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/acfbf162-0e20-5d3d-c72a-2ccfdf5a98ee.png)
[※出典：Tutorial: Getting started with Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-gs.html)

# 実践
[チュートリアル: Amazon EMR の開始方法](https://docs.aws.amazon.com/ja_jp/emr/latest/ManagementGuide/emr-gs.html)を元に試します。

## 前提
### VPCのセットアップ
VPCとプライベートサブネットを設定します。
### Amazon EMR のセットアップ
#### Amazon EC2 のキーペアと Linux インスタンスを作成
::: note info
作成済みの場合は、スキップします。
:::

1. EC2のサイドバーで「キーペア」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/87034d86-8dad-bb35-d271-8695b92de969.png)

1. 「キーペアを作成する」をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/079c5b66-a999-7ae9-67c3-7b32860c8f9d.png)

1. 下記を入力し、「キーペアを作成」をクリックします。
    1. キーペア名
    1. キーペアタイプを選択（RSA or ED25519）
    1. プライベートキーファイル形式（.pem or .ppk）

## ステップ 1: Amazon EMR クラスターを計画して設定する
### Amazon EMR 用のストレージを準備する
1. S3バケットとフォルダを作成します。
- 今回は、インプットデータ、アウトプットデータ、コード配置用のバケット3種類を作成しました。

### Amazon EMR の入力データを使用してアプリケーションを準備する
#### 入力データを準備する
1. [food_establishment_data.zip](https://docs.aws.amazon.com/ja_jp/emr/latest/ManagementGuide/samples/food_establishment_data.zip)からデータをダウンロードします。
    - ワシントン州キング郡にある保健局の 2006～2020 年の検査結果の修正版
1. ファイルの中身を確認してみます。
    - ヘッダーが含まれていました。
    ```
    $ head food_establishment_data.csv           
    name,inspection_result,inspection_closed_business,violation_type,violation_points
    100 LB CLAM,Incomplete,FALSE,,0
    100 LB CLAM,Unsatisfactory,FALSE,BLUE,5
    100 LB CLAM,Unsatisfactory,FALSE,RED,5
    100 LB CLAM,Unsatisfactory,FALSE,RED,10
    100 LB CLAM,Unsatisfactory,FALSE,RED,5
    100 LB CLAM,Complete,FALSE,,0
    100 LB CLAM,Complete,FALSE,,0
    100 PERCENT NUTRICION,Unsatisfactory,FALSE,BLUE,5
    100 PERCENT NUTRICION,Unsatisfactory,FALSE,BLUE,5
    $
    ```

1. ファイルをアップロードします。
    - 71MB程度のファイルでした。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/f4434a4d-242b-d46a-b342-0418d4ad023c.png)


#### スクリプトを準備する
1. 下記のコードをファイル名「health_violations.py」として保存します。ロジックは下記の通り。
    - argparseを使って、入力パラメータからインプット/アウトプットのS3パスを取得
    - 指定されたS3パスのファイルを読み込みます。
    - Sparkセッションを作成後、violation_type='RED'を絞り込み、カウントします。
    - 指定されたS3のパスに上書きします。

    ```
    import argparse
    
    from pyspark.sql import SparkSession
    
    def calculate_red_violations(data_source, output_uri):
        """
        Processes sample food establishment inspection data and queries the data to find the top 10 establishments
        with the most Red violations from 2006 to 2020.
    
        :param data_source: The URI of your food establishment data CSV, such as 's3://DOC-EXAMPLE-BUCKET/food-establishment-data.csv'.
        :param output_uri: The URI where output is written, such as 's3://DOC-EXAMPLE-BUCKET/restaurant_violation_results'.
        """
        with SparkSession.builder.appName("Calculate Red Health Violations").getOrCreate() as spark:
            # Load the restaurant violation CSV data
            if data_source is not None:
                restaurants_df = spark.read.option("header", "true").csv(data_source)
    
            # Create an in-memory DataFrame to query
            restaurants_df.createOrReplaceTempView("restaurant_violations")
    
            # Create a DataFrame of the top 10 restaurants with the most Red violations
            top_red_violation_restaurants = spark.sql("""SELECT name, count(*) AS total_red_violations 
              FROM restaurant_violations 
              WHERE violation_type = 'RED' 
              GROUP BY name 
              ORDER BY total_red_violations DESC LIMIT 10""")
    
            # Write the results to the specified output URI
            top_red_violation_restaurants.write.option("header", "true").mode("overwrite").csv(output_uri)
    
    if __name__ == "__main__":
        parser = argparse.ArgumentParser()
        parser.add_argument(
            '--data_source', help="The URI for you CSV restaurant data, like an S3 bucket location.")
        parser.add_argument(
            '--output_uri', help="The URI where output is saved, like an S3 bucket location.")
        args = parser.parse_args()
    
        calculate_red_violations(args.data_source, args.output_uri)
    			
    ```

1. コード用のS3バケットにアップロードします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/a7fde3b8-aa6d-c3ca-da6f-9af4927f7af9.png)

### Amazon EMR クラスターを起動する
1. マネコンのEMRで「クラスターを作成」をクリックします。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/3a64758a-c13c-dfc9-2f70-a851db4c6365.png)


1. 下記を入力して、「クラスターを作成」をクリックします。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/c81325de-32b0-de04-8ed9-3a77ffa7af11.png)
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/274e323b-96ee-9cd2-d228-0df3b2f132e1.png)

1. しばらくすると「待機中」になります。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/2e5e73ca-57ab-e107-1593-e572384dc2e5.png)

## ステップ 2: Amazon EMR クラスターを管理する
### Amazon EMR に作業を送信する
1. 作成したクラスターを選択します。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/2965b02e-3e5b-3ccd-17d0-ad641810d1cb.png)

1. 「ステップ」タブをクリックし、「ステップの追加」をクリックします。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/1ad834b3-38f1-747a-f65a-b4c7fcbe3e89.png)

1. 下記を入力し「追加」をクリックします。
    - ステップタイプ：Sparkアプリケーションを選択
    - 名前：Sparkアプリケーション（デフォルト）
    - デプロイモード：クラスタ
    - Spark-submitオプション：未入力
    - アプリケーションの場所：コードを指定
    - 引数：下記を指定（バケット名やファイル名は適宜変更）
    - 失敗時の操作：次へ
    ```
    --data_source s3://DOC-EXAMPLE-BUCKET/food_establishment_data.csv
    --output_uri s3://DOC-EXAMPLE-BUCKET/myOutputFolder						
    ```
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/df68d40c-64d1-0da6-7330-e405d0892698.png)


1. 追加直後は、「保留中」になります。
    ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/84538ff7-6790-c9ca-40a0-065942182000.png)

1. しばらくすると、「実行中」に変わります。
   ![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/78470442-db35-6e41-d4e4-2c99c6e04a1b.png)

 
1. 処理が終了すると「完了」に変わります。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/42629aeb-882d-b6ae-7820-8d71fd5305df.png)


### 結果を表示する
#### データの確認
1. S3の出力パスを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/a4c6ad79-bfb2-8156-2afa-5d1425157a37.png)

1. S3 SELECTを実行し、データを確認します。
    - 赤の違反が多い上位10施設がリストされます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/2d98f76a-2043-4ceb-a4c4-7838f35cc4f6.png)

#### ジョブの実行の確認
1. アプリケーションの履歴をクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/627ba64d-b357-e027-aca7-2a4e9f05c2aa.png)

1. Spark History Serverが起動されます。
1. App IDをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/85f03359-03fd-0ce2-cb6f-a9ccd034c8f2.png)

1. 直近のJob IDのDescriptionをクリックします。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/43e893ae-8c71-7669-97ee-93abb0c0879f.png)

1. Stageを確認します。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/dcca442e-1748-3b39-4012-b00efb554e8e.png)

1. メトリクスも確認できます。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/206276/473fe933-b60e-9faa-d80d-db3cd8e2c05b.png)

### (オプション) 実行中の Amazon EMR クラスターに接続する
1. セキュリティグループのインバウンドルールにSSHを追加します。
1. SSHでName Nodeに接続します。
```
$ ssh -i ~/.ssh/emr-key.pem hadoop@xxxxxxxxxx.ap-northeast-1.compute.amazonaws.com
Last login: Sun Jan 22 13:50:56 2023

       __|  __|_  )
       _|  (     /   Amazon Linux 2 AMI
      ___|\___|___|

https://aws.amazon.com/amazon-linux-2/
12 package(s) needed for security, out of 14 available
Run "sudo yum update" to apply all updates.
                                                                    
EEEEEEEEEEEEEEEEEEEE MMMMMMMM           MMMMMMMM RRRRRRRRRRRRRRR    
E::::::::::::::::::E M:::::::M         M:::::::M R::::::::::::::R   
EE:::::EEEEEEEEE:::E M::::::::M       M::::::::M R:::::RRRRRR:::::R 
  E::::E       EEEEE M:::::::::M     M:::::::::M RR::::R      R::::R
  E::::E             M::::::M:::M   M:::M::::::M   R:::R      R::::R
  E:::::EEEEEEEEEE   M:::::M M:::M M:::M M:::::M   R:::RRRRRR:::::R 
  E::::::::::::::E   M:::::M  M:::M:::M  M:::::M   R:::::::::::RR   
  E:::::EEEEEEEEEE   M:::::M   M:::::M   M:::::M   R:::RRRRRR::::R  
  E::::E             M:::::M    M:::M    M:::::M   R:::R      R::::R
  E::::E       EEEEE M:::::M     MMM     M:::::M   R:::R      R::::R
EE:::::EEEEEEEE::::E M:::::M             M:::::M   R:::R      R::::R
E::::::::::::::::::E M:::::M             M:::::M RR::::R      R::::R
EEEEEEEEEEEEEEEEEEEE MMMMMMM             MMMMMMM RRRRRRR      RRRRRR
                                                                    
[hadoop@ip-XXX-XX-XX-XX ~]$ 
```

# 考察
簡単にクラスタが起動できました。History Serverなども自動でセットアップされるのはいいですね。


# 参考
https://qiita.com/zumax/items/4b641f49616e03bd6c76

https://docs.aws.amazon.com/ja_jp/emr/latest/ManagementGuide/emr-gs.html#emr-gs-next-steps