import pandas as pd import numpy as np from datetime import datetime, timedelta # 注文データ生成（1000件） np.random.seed(42) n_orders = 1000 orders = pd.DataFrame({ ‘order_id’: range(1001, 1001 + n_orders), ‘customer_id’: np.random.randint(101, 201, n_orders), ‘product_id’: np.random.randint(2001, 2051, n_orders), ‘quantity’: np.random.randint(1, 5, n_orders), ‘unit_price’: np.random.choice([1500, 8000, 12000, 35000, 98000], n_orders), ‘order_date’: [ (datetime(2024, 1, 1) + timedelta(days=np.random.randint(0, 90))).strftime(‘%Y-%m-%d’) for _ in range(n_orders) ] }) orders.to_csv(‘orders.csv’, index=False) print(f”✅ orders.csv を生成しました（{n_orders}件）”)

# AWS CLI でバケット作成 aws s3 mb s3://my-ecommerce-data-lake-12345 –region ap-northeast-1 # フォルダ構造を作成 aws s3api put-object –bucket my-ecommerce-data-lake-12345 –key raw/ aws s3api put-object –bucket my-ecommerce-data-lake-12345 –key processed/ aws s3api put-object –bucket my-ecommerce-data-lake-12345 –key curated/

# Raw層にCSVをアップロード aws s3 cp orders.csv s3://my-ecommerce-data-lake-12345/raw/orders/ aws s3 cp customers.csv s3://my-ecommerce-data-lake-12345/raw/customers/ aws s3 cp products.csv s3://my-ecommerce-data-lake-12345/raw/products/ # パーティション分割（日付別）でアップロード aws s3 cp orders.csv s3://my-ecommerce-data-lake-12345/raw/orders/year=2024/month=01/day=15/

{ “Rules”: [ { “Id”: “MoveToIA”, “Status”: “Enabled”, “Transitions”: [ { “Days”: 30, “StorageClass”: “STANDARD_IA” }, { “Days”: 90, “StorageClass”: “GLACIER_IR” }, { “Days”: 365, “StorageClass”: “DEEP_ARCHIVE” } ] } ] }

# Crawlerを実行 aws glue start-crawler –name ecommerce-raw-crawler # 実行状態を確認 aws glue get-crawler –name ecommerce-raw-crawler –query ‘Crawler.State’

# Glue ETL Job（PySpark） import sys from awsglue.transforms import * from awsglue.utils import getResolvedOptions from pyspark.context import SparkContext from awsglue.context import GlueContext from awsglue.job import Job from pyspark.sql.functions import * # 初期化 args = getResolvedOptions(sys.argv, [‘JOB_NAME’]) sc = SparkContext() glueContext = GlueContext(sc) spark = glueContext.spark_session job = Job(glueContext) job.init(args[‘JOB_NAME’], args) # 1. Raw層からデータ読み込み orders_df = spark.read.format(“csv”) \ .option(“header”, “true”) \ .option(“inferSchema”, “true”) \ .load(“s3://my-ecommerce-data-lake-12345/raw/orders/”) customers_df = spark.read.format(“csv”) \ .option(“header”, “true”) \ .option(“inferSchema”, “true”) \ .load(“s3://my-ecommerce-data-lake-12345/raw/customers/”) products_df = spark.read.format(“csv”) \ .option(“header”, “true”) \ .option(“inferSchema”, “true”) \ .load(“s3://my-ecommerce-data-lake-12345/raw/products/”) # 2. データクレンジング orders_df = orders_df.na.drop() # 欠損値を除去 # 合計金額の計算 orders_df = orders_df.withColumn( “total_amount”, col(“quantity”) * col(“unit_price”) ) # 3. テーブル結合 orders_enriched = orders_df \ .join(customers_df, on=”customer_id”, how=”left”) \ .join(products_df, on=”product_id”, how=”left”) # 4. Processed層に保存（Parquet形式） orders_enriched.write \ .mode(“overwrite”) \ .partitionBy(“order_date”) \ .parquet(“s3://my-ecommerce-data-lake-12345/processed/orders_enriched/”) job.commit()

— Redshiftに接続してテーブル作成 CREATE TABLE orders_fact ( order_id INT, customer_id INT, product_id INT, quantity INT, unit_price DECIMAL(10,2), total_amount DECIMAL(10,2), order_date DATE, order_timestamp TIMESTAMP, customer_name VARCHAR(100), customer_email VARCHAR(100), customer_age INT, customer_gender VARCHAR(10), customer_prefecture VARCHAR(50), product_name VARCHAR(200), category VARCHAR(100), product_price DECIMAL(10,2) ) DISTKEY(customer_id) SORTKEY(order_date);

— COPYコマンドでS3からロード COPY orders_fact FROM ‘s3://my-ecommerce-data-lake-12345/processed/orders_enriched/’ IAM_ROLE ‘arn:aws:iam::123456789012:role/RedshiftCopyRole’ FORMAT AS PARQUET;

— データ件数と売上合計を確認 SELECT COUNT(*) AS total_orders, SUM(total_amount) AS total_sales, AVG(total_amount) AS avg_order_value FROM orders_fact; — 日別売上 SELECT order_date, COUNT(*) AS order_count, SUM(total_amount) AS daily_sales FROM orders_fact GROUP BY order_date ORDER BY order_date DESC LIMIT 10;

— Athenaで外部テーブルを作成 CREATE EXTERNAL TABLE IF NOT EXISTS orders_enriched ( order_id INT, customer_id INT, product_id INT, quantity INT, unit_price DECIMAL(10,2), total_amount DECIMAL(10,2), order_timestamp TIMESTAMP, customer_name STRING, customer_age INT, customer_gender STRING, customer_prefecture STRING, product_name STRING, category STRING ) PARTITIONED BY (order_date DATE) STORED AS PARQUET LOCATION ‘s3://my-ecommerce-data-lake-12345/processed/orders_enriched/’; — パーティションを認識させる MSCK REPAIR TABLE orders_enriched;

— カテゴリ別売上ランキング SELECT category, COUNT(*) AS order_count, SUM(total_amount) AS total_sales, AVG(total_amount) AS avg_order_value FROM orders_enriched WHERE order_date >= DATE ‘2024-01-01’ GROUP BY category ORDER BY total_sales DESC; — 都道府県別の顧客分析 SELECT customer_prefecture, COUNT(DISTINCT customer_id) AS unique_customers, COUNT(*) AS order_count, SUM(total_amount) AS total_sales FROM orders_enriched GROUP BY customer_prefecture ORDER BY total_sales DESC LIMIT 10;