【AWS】[Amazon EMR(Hadoop/Spark)]を使ったデータエンジニアリング【ハンズオン】

# **Amazon EMR を活用したデータエンジニアリング**

昨今、ITエンジニアのスキルは多様化しています。その中で最も顕著な分野とされるのが、データ処理の分野です。通信量の増加により、データ処理の需要は年々右肩上がりです。
そのため、データエンジニアリングが専門外のエンジニアでも、将来的なデータ処理量の増加を考えてシステム設計をしないといけなくなりました。
また、プロジェクトによっては、データ処理を専門に担当する人も増えてきているように思えます。大規模なバッチ処理を行うプロジェクトも多くなってきています。

以上のことから、備忘録もかねて、データエンジニアリングをハンズオンで行うための記事を上げました。

なお、このハンズオンの記事には、AWSソリューションアーキテクトアソシエイト以上の知識と経験がある方に向けています。

まずは概念を把握し、ハンズオンに進んでいきましょう。いきなりハンズオンにはいかないように！（無駄な料金がかかってしまいます）
※最低限の構成でハンズオンを実施していますが、少し料金がかかります

## **1. Amazon EMRとは？**
Amazon EMR（Elastic MapReduce）は、AWS 上で動作するビッグデータ処理のためのマネージドクラスタープラットフォームです。Apache Hadoop のエコシステムを活用し、大規模なデータ処理や機械学習のトレーニングを効率的に行うことができます。

EMR は特に **Apache Spark** を中心に利用されることが多く、データのスケーラブルな処理やストリーム処理が可能です。また、AWS の他のサービスと密接に統合されており、データの保存や管理が容易になります。また、実務上では、Apache Sparkを単体で使うことはなく、**Hadoopはストレージ & リソース管理、Sparkはデータ処理のエンジンとして機能する**のが一般的です。つまり、互いのカバー範囲を補完しあっているといえます。

HadoopとSparkの関係性を分かりやすく整理した表を作成しました。

### **HadoopとSparkの比較表**

| 項目             | **Hadoop（MapReduce）** | **Apache Spark** |
|----------------|----------------------|----------------|
| **目的** | 分散バッチ処理 | 分散データ処理（バッチ & ストリーミング） |
| **データ処理方式** | **バッチ処理**（MapReduce） | **バッチ + ストリーミング処理**（リアルタイム処理） |
| **スピード** | ディスクI/Oベースの処理で遅い | メモリベースの処理で高速 |
| **データストレージ** | HDFS（Hadoop Distributed File System） | HDFS, S3, Cassandra, HBase, JDBCなど |
| **プログラミングモデル** | Java（MapReduce） | Scala, Java, Python, R |
| **リアルタイム処理** | 不向き | **Spark Streamingで対応可** |
| **機械学習** | Mahout（やや遅い） | **MLlib（高速な機械学習ライブラリ）** |
| **適用範囲** | 大規模バッチ処理 | **バッチ + リアルタイム分析、機械学習、データ分析** |
| **使いやすさ** | コードが複雑 | **シンプルなAPI（RDD, DataFrame, Dataset）** |
| **ストレージ依存** | **HDFS必須** | **HDFS以外のストレージとも連携可能** |
| **エコシステムとの連携** | Hive, HBase, Pig, Oozie | Hive, HBase, Cassandra, Kafka |
| **耐障害性** | 高い（HDFSのレプリケーション） | 高い（RDDの耐障害性） |

---

### **HadoopとSparkの関係性**
- **SparkはHadoopの代替ではなく、補完関係にある**
  - SparkはHadoopのHDFSをデータストレージとして使用できる
  - SparkはHadoop YARNをリソース管理として利用できる
  - HadoopのMapReduceの代わりにSparkを利用すると処理が高速化される
- **Hadoop（MapReduce）はバッチ処理向きだが、Sparkはバッチ & リアルタイム処理が可能**
- **最近のビッグデータ処理ではHadoop + Sparkの組み合わせが一般的**

---

### **役割**
| 目的 | 推奨 |
|------|------|
| 大規模なバッチ処理（ログ解析、レポート生成） | **Hadoop（MapReduce）** |
| 高速データ処理 & リアルタイム分析 | **Spark** |
| データレイクとしての分散ストレージ | **HDFS（Hadoop）** |
| 機械学習・AIモデルのトレーニング | **Spark MLlib** |
| ストリーミングデータの処理（IoT, SNSデータ） | **Spark Streaming** |

といえると思います。

## **2. なぜ Amazon EMR を学ぶべきなのか？**
フリーランスとしても、大型案件のシステム設計の面接で頻繁に「**大量のデータをどのように処理するか？**」という質問を、しばしば受けます。その時、どの企業も何らかのビッグデータツールを導入していることに気が付きました。

データエンジニアリングの分野では、Python や SQL の知識だけでなく、**ビッグデータフレームワークを扱えるスキル** が求められています。
そのため、EMR を活用することで、大規模データの処理や ETL パイプラインの構築など、より高度なスキルを身につけることができると思います。

## **3. Amazon EMR の特徴**
### **3.1. EMR のクラスタ管理**
EMR は **YARN（Yet Another Resource Negotiator）** を使用し、クラスタのリソース管理を行います。

`補足: なお、YARNとは、クラスタリソースを管理するための仕組みであり、「リソースマネージャー」 と 「ノードマネージャー」 で構成されています。`

EMRのクラスタには、以下の 3 つのノードが存在します。

1. **プライマリノード（マスターノード）**
   - クラスタ全体を管理し、タスクを分配
   - クラスタの状態を監視

2. **コアノード（ワーカーノード）**
   - 実際にジョブを処理し、HDFS にデータを保存

3. **タスクノード（オプション）**
   - ワーカーノードと同じく処理を行うが、HDFS にはデータを保存しない

### **3.2. 他の AWS サービスとの統合**
EMR の大きな強みは、AWS の以下のサービスと連携できる点です。

- **Amazon S3**：データの永続的なストレージとして利用。EMRFS（EMR File System）を通じて S3 に保存・読み取りが可能
- **Amazon EC2**：クラスタのワーカーノードをプロビジョニング
- **Amazon VPC**：安全なネットワーク環境を構築

## **4. EMR のスケーリングとパフォーマンス管理**
EMR はスケーラブルなクラスタ管理が可能で、手動または **EMR マネージドスケーリング** によって、ジョブの負荷に応じたスケールアップ・ダウンを自動化できます。

例えば、ピーク時にはノードを増やし、アイドル時にはコスト削減のために縮小することができます。これは、企業にとって非常に大きなメリットとなります。もちろん、機会損失とコスト削減がクラウドコンピューティングを使用する最大の目的なので、外せないメリットにはなります。

## **5. Amazon EMR を活用した実践例**
では、ここからハンズオンに移りましょう！AWS CLIでも実践は可能ですが、GUIで実施する際のスクリーンショットを用いて、ハンズオンで紹介します。

まず、ハンズオンの手順としては以下の通り。

### **5.1. EMR クラスタの作成**
1. **S3 にデータを保存するためのバケットを作成**
- 「バケットを作成」をクリック
![スクリーンショット 2025-01-04 151709.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/c013923c-f429-3e9c-aa01-c603c52a952c.png)
- 画面の通りにオプションの選択・入力をし、バケットを作成
![スクリーンショット 2025-01-04 151801.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/6cda7a4e-0977-95df-62f9-62fb1b3b6983.png)
![スクリーンショット 2025-01-04 151818.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/c9e98607-06f4-d23a-c77c-95899709603e.png)
![スクリーンショット 2025-01-04 151855.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/4a07273e-1c2e-a991-410a-e246268fde11.png)
- 画像の通りにバケットの中に以下のフォルダーツリー(例)を構成
```
# s3://<yourname-mmdd>

labor-force-in-Hokkaido
│── input/                 # 元データを格納するディレクトリ
│   │── 01.csv             # 労働力データのCSVファイル
│      
│── output/                # 処理後のデータを格納するディレクトリ
│   ├── _SUCCESS            # 処理成功のマーカー
│   ├── _SUCCESS.crc        # CRC チェックサムファイル
│   ├── .part-00000-xxxxxx.snappy.parquet.crc  # データパーティションのチェックサム
│   ├── part-00000-xxxxxx.snappy.parquet       # Parquet 形式の出力データ
│
│── logs/                  # ジョブのログデータ
│   ├── some.log/          # EMR の実行ログ
│
│── main.py                # Sparkのスクリプト
```

![スクリーンショット 2025-01-04 151917.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/33373f26-027a-ba87-9178-24265eed102b.png)
![スクリーンショット 2025-01-04 152146.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/189d68e9-7ea4-ab00-54bc-f2559708bb95.png)
![スクリーンショット 2025-01-04 152249.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/4e7cb36e-152a-d190-8a74-6b3e72c189f8.png)


2. **VPC（仮想ネットワーク）を作成**
- 画像の通りにVPCの作成
![スクリーンショット 2025-01-04 152451.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/26140394-bc3c-55d5-228e-1014e14b90c6.png)
![スクリーンショット 2025-01-04 152615.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/95567f33-cb32-9318-2a11-ed597d9fe1f6.png)


3. **EMR クラスタを起動**
- 画像の通りにEMRクラスタを作成し、起動
![スクリーンショット 2025-01-04 152731.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/244b6c19-1fad-4c02-cc22-0541dc5c199e.png)
- 最低限HadoopとSparkのみを選択
![スクリーンショット 2025-01-04 152832.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/8b687cd3-761c-986a-61e8-23edd4073e71.png)
- クラスター設定にて、ユニフォームインスタンスグループを選択し、プライマリとコアのインスタンスタイプをm5.largeに設定
![スクリーンショット 2025-01-04 152941.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/e44701d0-b54f-61dc-3e67-7c565faabaae.png)
- インスタンスグループは不要なので、削除すること
![スクリーンショット 2025-01-04 152956.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/bb060917-8ef0-6240-e332-f49afecdee2d.png)
![スクリーンショット 2025-01-04 153003.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/1f33baae-8b5e-d695-9e86-62afc354caab.png)
![スクリーンショット 2025-01-04 153025.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/9fe1bd66-50be-8caa-84ca-5a6090720728.png)
![スクリーンショット 2025-01-04 153046.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/6f6ebe80-7d2c-7fa2-a612-f9f0201c916c.png)
![スクリーンショット 2025-01-04 153056.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/fe26e7df-bd01-0290-4094-f7935f653a77.png)
![スクリーンショット 2025-01-04 153108.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/71a4a104-e800-19cf-55fd-210c15603f8d.png)
** アイドル時間はハンズオンがちょうど終わる時間を逆算して設定してください **
![スクリーンショット 2025-01-04 153153.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/ed0b227c-2b43-3d55-2ad7-ac0aab25f116.png)
- ログの配置場所を、先ほど作ったBucketの中のフォルダーに設定します
![スクリーンショット 2025-01-04 153353.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/fbad28f4-ed4a-81f6-9ade-822d3d8ac313.png)
![スクリーンショット 2025-01-04 153506.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/2d30038c-c1fb-8ba1-c0e3-b25c1b9c2f5d.png)
- EC2キーペアは事前に取得したものを使ってください
![スクリーンショット 2025-01-04 153534.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/4cedca0e-40d8-fd27-dfcd-d6fc0c40f3f2.png)
- EMRサービスロールは新規で作成し、デフォルト設定で
![スクリーンショット 2025-01-04 153656.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/db07c98f-146a-4c8b-a6cd-b7ad85d4598e.png)
- EMR EC2インスタンスプロファイルを作成(!重要!: バケットアクセスを読み取りおよび書き込みのアクセス権を持つに設定することを推奨)
![スクリーンショット 2025-01-04 153726.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/9983f1d0-3238-1b58-5707-881b3585c085.png)
- しばらく待つと。(経験則ですが、だいたい5分以内には作成が完了します)
![スクリーンショット 2025-01-04 154127.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/cfa8dd49-dd68-4dab-5eae-4c98104bde4b.png)


これで、セットアップを完了！

### **5.2. EMR での Spark の実行**
1. **CSV データを S3 からロード**
- 以下のリンクからCSVをダウンロードし、任意の名前にリネームしてください(画像の例では、test_for_emr.csvにしています)
[北海道オープンデータポータル](https://www.harp.lg.jp/opendata/dataset/256/resource/635/%E7%AC%AC1%E8%A1%A8_%EF%BC%88%E7%B7%8F%E5%90%88%EF%BC%89%E6%8C%AF%E8%88%88%E5%B1%80%E3%83%BB%E5%B8%82%E5%8C%BA%E7%94%BA%E6%9D%91%E5%88%A5%E7%94%A3%E6%A5%AD%E5%A4%A7%E5%88%86%E9%A1%9E%E5%88%A5%E4%BA%8B%E6%A5%AD%E6%89%80%E6%95%B0%E3%80%81%E7%94%B7%E5%A5%B3%E5%88%A5%E5%BE%93%E6%A5%AD%E8%80%85%E6%95%B0.csv)
![スクリーンショット 2025-01-04 154439.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/ee047aee-d754-cca1-92a3-20cb6680dcbd.png)


2. ステップ作成前準備
スクリプトをステップで実行できるように配置します。
ステップとは、EMRクラスタ上で実行される一連のジョブやスクリプトのことです。ステップを利用することで、EMR クラスタ上で 自動的にデータ処理タスク を実行できます。

以下のようなステップを作成するとします。
**データソースのETL処理** -> **Parquet 形式で S3 に保存**

この処理をスクリプトを用いて、実施します。

- スクリプト作成
主にETL処理を施しているスクリプトです。

引数から渡されたデータソースを処理します。データセット上で、日本のオープンデータ特有の邪魔な列の削除や日本語名のカラムが扱いづらいので、pysparkのモジュールで変換などを行います。必要なライブラリは随時インストールしてください。
また、スクリプトの最後には、引数から指定されるURI内に、ETL処理を施したデータセットをParquet形式で保存します。
スクリプトは以下の通り。
```main.py
from pyspark.sql import SparkSession
from pyspark.sql.functions import col
import argparse


def transform_data(data_source: str, output_uri: str) -> None:
    """
    Transforms inspection data from a CSV file, removes unwanted rows,
    sets the 3rd row as column names, filters out entries containing '計',
    and saves the results as Parquet.

    Args:
        data_source (str): The S3 URI or local path to the input CSV file.
        output_uri (str): The S3 URI or local path where the transformed data should be saved.
    """
    # Initialize Spark session
    with SparkSession.builder.appName("EMR Application").getOrCreate() as spark:
        # Load CSV file into a DataFrame (without using first row as header)
        df = (
            spark.read.option("header", False)  # Don't use first row as header
            .option("encoding", "shift_jis")
            .csv(data_source)
        )
        print("CSV file loaded.")

        # Assign an index to each row
        indexed_rdd = df.rdd.zipWithIndex()

        # Extract the 3rd row (index 2) as column names
        header_row = (
            indexed_rdd.filter(lambda row: row[1] == 2)
            .map(lambda row: row[0])
            .collect()[0]
        )
        print(f"New header extracted: {header_row}")

        # Remove the first six rows except for the 3rd (index 2)
        rows_to_remove = {0, 1, 3, 4, 5}
        filtered_rdd = indexed_rdd.filter(lambda row: row[1] not in rows_to_remove).map(
            lambda row: row[0]
        )

        # Convert back to DataFrame using new column names
        df = spark.createDataFrame(filtered_rdd, schema=header_row)
        print("Header applied and selected rows removed successfully.")

        # Remove rows containing "計" in the 市町村 (City or Town) column
        df = df.filter((col("市町村").isNotNull()) & (~col("市町村").contains("計")))
        print("Rows containing '計' removed successfully.")

        # Rename columns
        df = df.select(
            col("振興局").alias("county"),
            col("コード").alias("location_code"),
            col("市町村").alias("city_or_town"),
            col("区").alias("ward"),
            col("Ａ〜Ｒ全産業（Ｓ公務を除く）_事業所数").alias(
                "workplaces_in_all_industries"
            ),
            col("Ａ〜Ｒ全産業（Ｓ公務を除く）_従業者数").alias(
                "workers_in_all_industries"
            ),
            col("Ａ〜Ｒ全産業（Ｓ公務を除く）_従業者数_男(人)").alias(
                "male_workers_in_all_industries"
            ),
            col("Ａ〜Ｒ全産業（Ｓ公務を除く）_従業者数_女(人)").alias(
                "female_workers_in_all_industries"
            ),
            col("Ａ〜Ｂ農林漁業_事業所数").alias("workplaces_in_primary_industry"),
            col("Ａ〜Ｂ農林漁業_従業者数").alias("workers_in_primary_industry"),
            col("Ａ〜Ｂ農林漁業_従業者数_男(人)").alias(
                "male_workers_in_primary_industry"
            ),
            col("Ａ〜Ｂ農林漁業_従業者数_女(人)").alias(
                "female_workers_in_primary_industry"
            ),
        )

        # Register temporary SQL table
        df.createOrReplaceTempView("labor_force_in_Hokkaido")

        # SQL query to preview data
        transformed_df = spark.sql("""
            SELECT county, 
                SUM(workplaces_in_all_industries) AS total_workplaces, 
                SUM(workers_in_all_industries) AS total_workers, 
                SUM(male_workers_in_all_industries) AS total_male_workers, 
                SUM(female_workers_in_all_industries) AS total_female_workers, 
                SUM(workplaces_in_primary_industry) AS total_workplaces_primary_industries, 
                SUM(workers_in_primary_industry) AS total_workers_primary_industry
            FROM labor_force_in_Hokkaido
            GROUP BY county
        """)

        transformed_df.show(truncate=False)

        # Log the number of rows
        row_count = transformed_df.count()
        print(f"Number of rows in transformed data: {row_count}")

        # Save results in Parquet format
        transformed_df.write.mode("overwrite").parquet(output_uri)
        print(f"Data written to {output_uri}.")


if __name__ == "__main__":
    """
    Parses command-line arguments and executes the transform_data function.
    """
    parser = argparse.ArgumentParser(description="Transform inspection data.")
    parser.add_argument(
        "--data_source", required=True, help="Path to the input CSV file."
    )
    parser.add_argument(
        "--output_uri",
        required=True,
        help="Path to the output directory for Parquet files.",
    )

    args = parser.parse_args()

    # Execute the data transformation process
    transform_data(args.data_source, args.output_uri)
```

- ステップで実行するスクリプトをS3に収納する
![スクリーンショット 2025-01-04 154824.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/32078499-9450-7fb8-5b06-1d019ae36ed5.png)


3. **ステップ作成**
![スクリーンショット 2025-01-04 154523.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/23899f4f-ddea-c632-7f4c-6f8fcff717d9.png)
![スクリーンショット 2025-01-04 154537.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/b33223bd-c322-829f-92db-395c8b350c28.png)
![スクリーンショット 2025-01-04 154955.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/ad240f7e-899f-42b7-5e44-5fe3a3d890fb.png)
![スクリーンショット 2025-01-04 155020.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/cad18ede-0893-6caf-b62e-9e7b92ad54bd.png)
- ステップ作成完了を待ちます
![スクリーンショット 2025-01-04 155212.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/26504123-6840-cd8d-a645-230dba8ebb9b.png)
完成！



### **5.3. SSH で EMR クラスタに接続**
ちなみに、EMR のプライマリノードに SSH で接続し、手動で Spark スクリプトを実行できます。
そのためには、まずプライマリノードへのSSH接続を有効化してください。
- SSH接続の有効化
![スクリーンショット 2025-01-04 155550.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/9839b3e6-fa41-02fd-634d-88feb6032d27.png)
![スクリーンショット 2025-01-04 155622.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/e582f33e-d618-e023-f87f-434a1227e872.png)

- SSH接続
pemキーなどの場所を指定し、プライマリノードに接続します。
```sh
ssh -i my-key.pem hadoop@ec2-xx-xx-xx-xx.compute-1.amazonaws.com
```
![スクリーンショット 2025-01-04 160552.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/355c7f9e-43fc-cda7-8c9c-baec1f7be9b2.png)
- sparkコマンドで処理を実行
spark-submitコマンドを使って処理を実行します。
引数には、
--data_source **[S3バケットinput内のデータソースのパス]**
--output_uri **[S3バケットoutputフォルダーのパス]**
という風にオプションを入れます。
![スクリーンショット 2025-01-04 160957.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/82334275-1bf2-075c-46dd-a5dcf9a5480e.png)

- 実行結果の確認
![スクリーンショット 2025-01-04 161019.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/1557f83f-277f-6c7d-335a-9729a46b7660.png)

### **5.4. SSH で EMR クラスタのUIにアクセス**
- EMR UI アプリケーションのセットアップ
EMRのUIには、基本的にSSHトンネリングをセットアップし、アクセスするようにします。永続的なアプリケーションUIは、実務上特別な事由がない限りは選ばないと思います。アプリケーションのタグをクリックし、画像の通り選択してください。
![スクリーンショット 2025-01-04 161158.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/ed03b818-1083-c01f-ae51-e53c0d30e5e5.png)

- sshのトンネリングを実行
ローカルマシンの8157ポートを、ユーザー名hadoopでEMRクラスタの8088に転送するようにします。 
![スクリーンショット 2025-01-04 161440.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/e9cb50ce-e72d-29b8-d737-e31c47f76ad0.png)

- UI にアクセス
ブラウザでlocalhost:8157にアクセスし、HadoopのUIを確認
![スクリーンショット 2025-01-04 161520.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/3976101/475dd2ec-1128-1dd0-d916-1bb2e5733bc9.png)

よく見てみると、ノードの数（現在はプライマリとセカンダリのみ）だけ、アプリケーションIDが発行されています。


## **6. Amazon EMR の課題と対策**
### **6.1. ログのデバッグが難しい**
EMR は強力なツールですが、**ログのデバッグが非常に難しい** という課題があります。特に、ジョブの失敗時にエラーログの特定が難しいため、適切な監視ツールを活用することが重要です。
というのは、Amazon EMRを使うと、Spark/Hadoopのアプリケーションログ、EMRシステムログ、CloudWatchに送信されるログの３種類が生成されます。レイヤーごとに使う監視ツールを連携させないといけないのが難しい部分となります。

### **6.2. コストの最適化**
EMR はオンデマンドで利用できますが、**適切にスケーリングしないとコストが増加** します。コストを最適化するために、**スポットインスタンス** や **EMR マネージドスケーリング** を活用することを推奨します。

## **7. まとめ**
データエンジニアリングのスキルを向上させるために、Amazon EMR を活用して **大規模データの処理** を学びましょう。

### **✅ 本記事のポイント**
✔️ **EMR は AWS 上のビッグデータ処理のためのマネージドサービス**
✔️ **Apache Spark を活用してデータ処理が可能**
✔️ **Amazon S3, EC2, VPC との統合がスムーズ**
✔️ **スケーラブルなクラスタ管理が可能**
✔️ **ログのデバッグやコスト管理が課題となる**

なお、後日、HadoopとSparkを組み合わせてUIでどのように操作するかを解説する記事をアップしていきます。（余裕があれば）

**ぜひ、データエンジニアリングのスキルを強化していきましょう！**


## **参考文献・公式ドキュメント**
### **Amazon EMR**
1. **Amazon EMR 公式ドキュメント**  
   - [Amazon EMR](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-what-is-emr.html)  
   - AWSのEMRに関する公式ガイド。基本概念や構成方法、チュートリアルが記載されています

2. **AWS EMR Best Practices**  
   - [Amazon EMR Best Practices Guide](https://docs.aws.amazon.com/emr/latest/ManagementGuide/emr-best-practices.html)  
   - Amazon EMRを最適に利用するためのガイドライン

3. **Amazon EMR での Apache Spark の利用**  
   - [Running Spark on Amazon EMR](https://docs.aws.amazon.com/emr/latest/ReleaseGuide/emr-spark.html)  
   - Amazon EMR上でSparkを実行するための公式ガイド

4. **AWS Big Data Blog - Amazon EMR**  
   - [AWS Big Data Blog](https://aws.amazon.com/blogs/big-data/)  
   - Amazon EMRやビッグデータ関連の実践記事が掲載されているブログ

---

### **Hadoop**
5. **Apache Hadoop 公式ドキュメント**  
   - [Apache Hadoop](https://hadoop.apache.org/docs/stable/)  
   - Hadoopの基本構成やMapReduce、HDFSの詳細情報が掲載されています
   - 
6. **HDFS（Hadoop Distributed File System）**  
   - [HDFS Architecture Guide](https://hadoop.apache.org/docs/stable/hadoop-project-dist/hadoop-hdfs/HdfsDesign.html)  
   - HDFSのデータ保存、分散システム、レプリケーションに関する詳細な解説

7. **Hadoop vs. Sparkの違い**  
   - [Hadoop vs. Spark: Understanding the Differences](https://data-flair.training/blogs/hadoop-vs-spark/)  
   - HadoopのMapReduceとSparkの違いを詳しく解説した記事

---

### **Apache Spark**
8. **Apache Spark 公式ドキュメント**  
   - [Apache Spark](https://spark.apache.org/docs/latest/)  
   - Sparkの基本概念、RDD、DataFrame、Spark Streaming、MLlibに関する詳細な公式情報

9. **Spark Performance Tuning Guide**  
   - [Tuning Spark](https://spark.apache.org/docs/latest/tuning.html)  
   - Sparkのパフォーマンスを最適化するための公式ガイド

10. **SparkとHadoopの違いに関する詳細な記事**  
   - [Spark vs Hadoop: Key Differences](https://www.databricks.com/blog/2014/10/10/spark-hadoop-mapreduce-differences.html)  
   - DatabricksによるHadoop MapReduceとSparkの違いの解説

---

### **AWS CLI & S3**
11. **AWS CLI コマンドリファレンス**  
   - [AWS CLI Command Reference](https://docs.aws.amazon.com/cli/latest/reference/)  
   - AWS CLIを使ってEMRやS3を操作する際のコマンドリスト

12. **Amazon S3 の公式ドキュメント**  
   - [Amazon S3](https://docs.aws.amazon.com/AmazonS3/latest/dev/Welcome.html)  
   - Amazon S3のデータ保存と管理についての公式ガイド

---

### **データセット**
13. **北海道オープンデータポータル**（本記事のデータソース）  
   - [北海道オープンデータポータル](https://www.harp.lg.jp/opendata/)  
   - 労働力データなどの公的なオープンデータセットを提供

14. **UCI Machine Learning Repository**（追加データ参考）  
   - [UCI ML Repository](https://archive.ics.uci.edu/ml/index.php)  
   - 機械学習用のオープンデータセットを多数提供

---

### **追加資料（実務的な記事）**
15. **AWS EMRを活用したデータ処理の実践**  
   - [AWS Black Belt Online Seminar – Amazon EMR](https://aws.amazon.com/jp/about-aws/whats-new/2019/10/aws-black-belt-online-seminar-amazon-emr/)  
   - AWSの公式ウェビナーでEMRの活用事例を学べる

16. **Hadoop/Sparkを活用したビッグデータ分析の実践**  
   - [Big Data Processing with Hadoop and Spark](https://towardsdatascience.com/apache-spark-vs-hadoop-mapreduce-differences-and-advantages-5fba829ed6b7)  
   - 実際のビッグデータ処理ワークフローを解説

17. **AWS EMRのコスト最適化ガイド**  
   - [How to Optimize Amazon EMR Costs](https://aws.amazon.com/blogs/big-data/how-to-optimize-amazon-emr-costs/)  
   - EMRのコスト管理とスポットインスタンス活用方法