☁️ 1. AWS EMRとは？

1-1. EMR（Elastic MapReduce）の概要

AWS EMRは、AWSが提供するマネージド型のビッグデータ処理サービスです。
Apache Spark、Hadoop、Prestoなどの分散処理フレームワークを、簡単にクラスター上で実行できます。

1-2. EMRを使うメリット

1-3. EMRのアーキテクチャ

🔧 2. 事前準備

2-1. AWSアカウントの準備

EMRを使用するには、AWSアカウントが必要です。
まだアカウントがない場合は、AWS公式サイトから作成してください。

2-2. S3バケットの作成

EMRで処理するデータは、S3（Simple Storage Service）に保存します。
まず、データ保存用のS3バケットを作成しましょう。

1. AWSコンソールにログイン
2. 「S3」を検索してS3コンソールを開く
3. 「バケットを作成」をクリック
4. バケット名を入力（例：my-spark-data-bucket）
5. リージョンを選択（例：ap-northeast-1 東京）
6. 「バケットを作成」をクリック

2-3. サンプルデータのアップロード

# ローカルでサンプルCSVを作成
# sales_data.csv

date,product,category,amount,region
2024-01-01,商品A,食品,1000,東京
2024-01-01,商品B,電化製品,50000,大阪
2024-01-02,商品C,衣料品,15000,東京
2024-01-02,商品D,食品,2500,名古屋
2024-01-03,商品E,電化製品,80000,大阪

作成したCSVファイルを、S3バケットにアップロードします。

1. S3コンソールで作成したバケットを開く
2. 「アップロード」をクリック
3. CSVファイルをドラッグ＆ドロップ
4. 「アップロード」をクリック

アップロード先：
s3://my-spark-data-bucket/input/sales_data.csv

🚀 3. EMRクラスターの作成

3-1. EMRコンソールからクラスター作成

1. AWSコンソールで「EMR」を検索してEMRコンソールを開く
2. 「クラスターを作成」をクリック
3. クラスター名を入力（例：my-spark-cluster）
4. EMRリリースを選択（最新版を推奨：emr-7.0.0など）
5. アプリケーションで「Spark」を選択

3-2. インスタンス設定

3-3. その他の設定

# キーペア設定
EC2キーペア：既存のキーペアを選択、または新規作成
（SSHでマスターノードに接続する場合に必要）

# ログ設定
S3バケット：s3://my-spark-data-bucket/logs/

# セキュリティ設定
・EMRマネージドセキュリティグループを使用（デフォルト）
・必要に応じてSSHアクセス（ポート22）を許可

3-4. クラスター作成の完了

設定が完了したら「クラスターを作成」をクリックします。
クラスターの起動には約10〜15分かかります。

クラスターのステータス：
Starting → Bootstrapping → Running ← この状態になればOK！

※「Waiting」と表示される場合もあります（ジョブ待機中）

📦 4. S3との連携

4-1. S3からデータを読み込む

EMRでは、S3のデータを直接読み込むことができます。
S3のパスはs3://バケット名/パスの形式で指定します。

# S3からCSVを読み込む

from pyspark.sql import SparkSession

spark = SparkSession.builder \
    .appName("EMR S3 Example") \
    .getOrCreate()

# S3からデータを読み込み
df = spark.read.csv(
    "s3://my-spark-data-bucket/input/sales_data.csv",
    header=True,
    inferSchema=True
)

df.show()

+----------+-------+--------+------+------+
|      date|product|category|amount|region|
+----------+-------+--------+------+------+
|2024-01-01|  商品A|    食品|  1000|  東京|
|2024-01-01|  商品B|電化製品| 50000|  大阪|
|2024-01-02|  商品C|  衣料品| 15000|  東京|
|2024-01-02|  商品D|    食品|  2500|名古屋|
|2024-01-03|  商品E|電化製品| 80000|  大阪|
+----------+-------+--------+------+------+

4-2. S3にデータを書き込む

# 集計結果をS3に保存

# カテゴリ別の売上集計
result = df.groupBy("category").sum("amount")

# Parquet形式でS3に保存（推奨）
result.write.parquet(
    "s3://my-spark-data-bucket/output/category_sales/",
    mode="overwrite"
)

print("S3への保存完了")

S3への保存完了

保存先：
s3://my-spark-data-bucket/output/category_sales/
├── part-00000-xxx.parquet
├── part-00001-xxx.parquet
└── _SUCCESS

▶️ 5. Sparkジョブの実行方法

5-1. EMR Notebooks（推奨）

EMR Notebooksは、Jupyter Notebookベースの対話的な開発環境です。
ローカルのJupyter Notebookと同じ感覚でSparkコードを実行できます。

1. EMRコンソールで「Notebooks」を選択
2. 「ノートブックを作成」をクリック
3. ノートブック名を入力
4. クラスターを選択（作成したクラスター）
5. 「ノートブックを作成」をクリック
6. ステータスが「Ready」になったら「開く」をクリック

5-2. spark-submitでジョブ投入

バッチ処理の場合は、spark-submitでPythonスクリプトを実行します。

# Pythonスクリプト（sales_analysis.py）をS3にアップロード

# sales_analysis.py の内容
from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("Sales Analysis").getOrCreate()

# S3から読み込み
df = spark.read.csv("s3://my-spark-data-bucket/input/sales_data.csv", 
                    header=True, inferSchema=True)

# 集計
result = df.groupBy("region").agg(
    F.sum("amount").alias("total_sales"),
    F.count("*").alias("order_count")
)

# S3に保存
result.write.parquet("s3://my-spark-data-bucket/output/region_sales/", 
                     mode="overwrite")

spark.stop()

5-3. EMRステップとしてジョブを追加

1. EMRコンソールでクラスターを選択
2. 「ステップ」タブを開く
3. 「ステップを追加」をクリック
4. ステップタイプ：「Spark アプリケーション」を選択
5. Spark-submit オプション：--deploy-mode cluster
6. アプリケーションの場所：s3://my-spark-data-bucket/scripts/sales_analysis.py
7. 「ステップを追加」をクリック

ステップのステータス：
Pending → Running → Completed

ログの確認先：
s3://my-spark-data-bucket/logs/

💰 6. クラスターの停止とコスト管理

6-1. クラスターの終了（重要！）

1. EMRコンソールでクラスターを選択
2. 「終了」をクリック
3. 確認ダイアログで「終了」をクリック
4. ステータスが「Terminated」になったことを確認

6-2. コストの目安

6-3. コスト削減のポイント

📝 STEP 23 のまとめ