# SparkSessionの初期化

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

# SparkSessionを作成（または既存のものを取得）
spark = SparkSession.builder \
    .appName("GroupByPractice") \
    .getOrCreate()

print("SparkSession準備完了")

# 基本的な売上データを作成

# データ：(商品名, カテゴリ, 売上金額)
sales_data = [
    ("商品A", "食品", 1000),
    ("商品B", "食品", 1500),
    ("商品C", "電化製品", 30000),
    ("商品D", "電化製品", 25000),
    ("商品E", "衣料品", 5000),
    ("商品F", "衣料品", 8000)
]

# DataFrameを作成（カラム名を指定）
df_sales = spark.createDataFrame(sales_data, ["product", "category", "sales"])

print("売上データ:")
df_sales.show()

# 月別・店舗別の売上データを作成

# データ：(月, 店舗, 売上金額)
monthly_data = [
    ("2024-01", "店舗A", 1000000),
    ("2024-01", "店舗B", 1500000),
    ("2024-01", "店舗C", 800000),
    ("2024-02", "店舗A", 1200000),
    ("2024-02", "店舗B", 1400000),
    ("2024-02", "店舗C", 900000),
    ("2024-03", "店舗A", 1100000),
    ("2024-03", "店舗B", 1600000),
    ("2024-03", "店舗C", 850000)
]

df_monthly = spark.createDataFrame(monthly_data, ["month", "store", "sales"])

print("月別・店舗別売上データ:")
df_monthly.show()

# ========================================
# STEP 11 サンプルデータ一括作成スクリプト
# ========================================

from pyspark.sql import SparkSession
from pyspark.sql import functions as F

spark = SparkSession.builder.appName("GroupByPractice").getOrCreate()

# 1. 基本的な売上データ
sales_data = [
    ("商品A", "食品", 1000), ("商品B", "食品", 1500),
    ("商品C", "電化製品", 30000), ("商品D", "電化製品", 25000),
    ("商品E", "衣料品", 5000), ("商品F", "衣料品", 8000)
]
df_sales = spark.createDataFrame(sales_data, ["product", "category", "sales"])

# 2. 月別・店舗別データ
monthly_data = [
    ("2024-01", "店舗A", 1000000), ("2024-01", "店舗B", 1500000), ("2024-01", "店舗C", 800000),
    ("2024-02", "店舗A", 1200000), ("2024-02", "店舗B", 1400000), ("2024-02", "店舗C", 900000),
    ("2024-03", "店舗A", 1100000), ("2024-03", "店舗B", 1600000), ("2024-03", "店舗C", 850000)
]
df_monthly = spark.createDataFrame(monthly_data, ["month", "store", "sales"])

# 3. 地域別・カテゴリ別データ（pivot用）
region_data = [
    ("東京", "食品", 500000), ("東京", "電化製品", 2000000), ("東京", "衣料品", 800000),
    ("大阪", "食品", 400000), ("大阪", "電化製品", 1800000), ("大阪", "衣料品", 700000),
    ("名古屋", "食品", 300000), ("名古屋", "電化製品", 1500000), ("名古屋", "衣料品", 600000)
]
df_region = spark.createDataFrame(region_data, ["region", "category", "sales"])

print("✅ サンプルデータを作成しました")
print("  - df_sales: 基本売上データ")
print("  - df_monthly: 月別・店舗別データ")
print("  - df_region: 地域別・カテゴリ別データ")

# カテゴリ別の売上合計を計算

# groupBy("category")：categoryの値でグループ化
# sum("sales")：各グループのsalesを合計
result = df_sales.groupBy("category").sum("sales")

print("カテゴリ別売上合計:")
result.show()

# カテゴリ別の商品数をカウント

# count()：各グループのレコード数をカウント
result = df_sales.groupBy("category").count()

print("カテゴリ別商品数:")
result.show()

# 月と店舗の両方でグループ化

# groupBy("month", "store")：月×店舗の組み合わせでグループ化
result = df_monthly.groupBy("month", "store").sum("sales")

# orderBy()：結果をソート（見やすくするため）
result = result.orderBy("month", "store")

print("月別・店舗別売上:")
result.show()

# 月別の売上合計（sum）

result = df_monthly.groupBy("month").sum("sales")
print("月別売上合計:")
result.orderBy("month").show()

# 店舗別の平均売上（avg）

result = df_monthly.groupBy("store").avg("sales")
print("店舗別平均売上:")
result.show()

# 店舗別の最高・最低売上（max, min）

# F.max()とF.min()を使用するためagg()を使う
result = df_monthly.groupBy("store").agg(
    F.max("sales").alias("最高売上"),
    F.min("sales").alias("最低売上")
)
print("店舗別の最高・最低売上:")
result.show()

# countの違いを確認するためNULLを含むデータを作成

data_with_null = [
    ("A", 100),
    ("A", None),  # NULLを含む
    ("A", 200),
    ("B", 150),
    ("B", None)   # NULLを含む
]
df_null = spark.createDataFrame(data_with_null, ["group", "value"])

# count("*")：NULLを含めた全件数
# count("value")：NULL以外の件数
result = df_null.groupBy("group").agg(
    F.count("*").alias("全件数"),
    F.count("value").alias("NULL以外")
)
result.show()

# 月別の詳細統計を計算

result = df_monthly.groupBy("month").agg(
    # count("*")：レコード数（店舗数）
    F.count("*").alias("店舗数"),
    
    # sum("sales")：売上合計
    F.sum("sales").alias("売上合計"),
    
    # avg("sales")：平均売上
    F.avg("sales").alias("平均売上"),
    
    # max("sales")：最高売上
    F.max("sales").alias("最高売上"),
    
    # min("sales")：最低売上
    F.min("sales").alias("最低売上")
)

print("月別詳細統計:")
result.orderBy("month").show()

# 計算式を含む集計

# 商品データに数量を追加
product_data = [
    ("食品", "商品A", 100, 1000),
    ("食品", "商品B", 150, 1500),
    ("電化製品", "商品C", 10, 50000),
    ("電化製品", "商品D", 5, 80000)
]
df_product = spark.createDataFrame(product_data, 
    ["category", "product", "quantity", "sales"])

# カテゴリ別の統計と単価計算
result = df_product.groupBy("category").agg(
    F.sum("quantity").alias("総数量"),
    F.sum("sales").alias("総売上"),
    
    # 計算式：総売上 ÷ 総数量 = 平均単価
    (F.sum("sales") / F.sum("quantity")).alias("平均単価")
)

print("カテゴリ別統計（単価計算付き）:")
result.show()

# 条件付き集計

result = df_monthly.groupBy("store").agg(
    # 全件数
    F.count("*").alias("全月数"),
    
    # 条件付きカウント：100万円以上の月数
    # when(条件, 1)：条件に合えば1、otherwise(0)：合わなければ0
    # sum()で合計すると条件に合った件数になる
    F.sum(F.when(F.col("sales") >= 1000000, 1).otherwise(0)).alias("100万以上"),
    
    # 条件付き合計：150万円以上の売上合計
    F.sum(F.when(F.col("sales") >= 1500000, F.col("sales")).otherwise(0)).alias("150万以上売上"),
    
    # 売上合計
    F.sum("sales").alias("売上合計")
)

print("条件付き集計結果:")
result.show()

# 店舗×月のクロス集計

# groupBy("store")：店舗を行にする
# pivot("month")：月を列にする
# sum("sales")：各セルに売上合計を表示
pivot_result = df_monthly.groupBy("store") \
    .pivot("month") \
    .sum("sales")

print("店舗×月のクロス集計:")
pivot_result.show()

# 値を指定してpivot（パフォーマンス向上）

# pivot()の第2引数に値のリストを渡す
pivot_result = df_monthly.groupBy("store") \
    .pivot("month", ["2024-01", "2024-02", "2024-03"]) \
    .agg(F.sum("sales"))

print("値を指定したpivot:")
pivot_result.show()

# 合計列を追加

pivot_result = df_monthly.groupBy("store") \
    .pivot("month", ["2024-01", "2024-02", "2024-03"]) \
    .agg(F.sum("sales"))

# withColumn()で合計列を追加
# F.col("列名")でカラムを参照
pivot_with_total = pivot_result.withColumn(
    "合計",
    F.col("2024-01") + F.col("2024-02") + F.col("2024-03")
)

print("合計列付きクロス集計:")
pivot_with_total.show()

# 地域×カテゴリのクロス集計

pivot_result = df_region.groupBy("region") \
    .pivot("category", ["食品", "電化製品", "衣料品"]) \
    .agg(F.sum("sales"))

# 合計列を追加
pivot_with_total = pivot_result.withColumn(
    "合計",
    F.col("食品") + F.col("電化製品") + F.col("衣料品")
)

print("地域×カテゴリ売上:")
pivot_with_total.show()

# ECサイトの注文データを作成

order_data = [
    ("2024-01-15", "店舗A", "食品", 50, 5000, "東京"),
    ("2024-01-15", "店舗A", "電化製品", 2, 80000, "東京"),
    ("2024-01-16", "店舗B", "衣料品", 10, 30000, "大阪"),
    ("2024-01-16", "店舗B", "食品", 30, 3000, "大阪"),
    ("2024-01-17", "店舗C", "電化製品", 5, 200000, "名古屋"),
    ("2024-01-17", "店舗A", "衣料品", 15, 45000, "東京"),
    ("2024-01-18", "店舗B", "食品", 40, 4000, "大阪"),
    ("2024-01-18", "店舗C", "電化製品", 3, 120000, "名古屋"),
    ("2024-01-19", "店舗A", "食品", 60, 6000, "東京"),
    ("2024-01-19", "店舗B", "衣料品", 20, 60000, "大阪")
]

df_order = spark.createDataFrame(order_data, 
    ["date", "store", "category", "quantity", "sales", "region"])

# 日付を型変換
df_order = df_order.withColumn("date", F.to_date("date", "yyyy-MM-dd"))

print("ECサイト注文データ:")
df_order.show()

# 店舗別の総合パフォーマンス分析

store_performance = df_order.groupBy("store", "region").agg(
    F.count("*").alias("取引件数"),
    F.sum("quantity").alias("総販売数"),
    F.sum("sales").alias("総売上"),
    F.avg("sales").alias("平均売上"),
    (F.sum("sales") / F.sum("quantity")).alias("平均単価")
).orderBy(F.desc("総売上"))

print("【分析1】店舗別パフォーマンス:")
store_performance.show(truncate=False)

# 地域×カテゴリのクロス集計

region_category = df_order.groupBy("region") \
    .pivot("category", ["食品", "電化製品", "衣料品"]) \
    .agg(F.sum("sales"))

# NULLを0に置換して合計を計算
region_category = region_category.fillna(0)
region_category = region_category.withColumn(
    "合計",
    F.col("食品") + F.col("電化製品") + F.col("衣料品")
)

print("【分析2】地域×カテゴリ売上:")
region_category.show()