# SparkSessionの初期化

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

# SparkSessionを作成
spark = SparkSession.builder \
    .appName("Spark SQL Basics") \
    .getOrCreate()

print("SparkSession準備完了")

# 売上データを作成

# データ：(商品, カテゴリ, 価格, 数量)
sales_data = [
    ("商品A", "食品", 1000, 10),
    ("商品B", "食品", 1500, 15),
    ("商品C", "電化製品", 50000, 2),
    ("商品D", "電化製品", 80000, 1),
    ("商品E", "衣料品", 5000, 5),
    ("商品F", "衣料品", 8000, 3)
]

df_sales = spark.createDataFrame(sales_data, ["product", "category", "price", "quantity"])

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

# 社員データを作成

# データ：(部門, 名前, 給与)
employees_data = [
    ("営業部", "田中", 300000),
    ("営業部", "鈴木", 350000),
    ("営業部", "佐藤", 320000),
    ("技術部", "山田", 400000),
    ("技術部", "伊藤", 450000),
    ("管理部", "渡辺", 380000)
]

df_employees = spark.createDataFrame(employees_data, ["department", "name", "salary"])

print("社員データ:")
df_employees.show()

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

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

spark = SparkSession.builder.appName("Spark SQL Basics").getOrCreate()

# 1. 売上データ
sales_data = [
    ("商品A", "食品", 1000, 10), ("商品B", "食品", 1500, 15),
    ("商品C", "電化製品", 50000, 2), ("商品D", "電化製品", 80000, 1),
    ("商品E", "衣料品", 5000, 5), ("商品F", "衣料品", 8000, 3)
]
df_sales = spark.createDataFrame(sales_data, ["product", "category", "price", "quantity"])

# 2. 社員データ
employees_data = [
    ("営業部", "田中", 300000), ("営業部", "鈴木", 350000), ("営業部", "佐藤", 320000),
    ("技術部", "山田", 400000), ("技術部", "伊藤", 450000), ("管理部", "渡辺", 380000)
]
df_employees = spark.createDataFrame(employees_data, ["department", "name", "salary"])

# 3. 月別売上データ
monthly_data = [
    ("2024-01", 1000000), ("2024-02", 1200000),
    ("2024-03", 1100000), ("2024-04", 1300000)
]
df_monthly = spark.createDataFrame(monthly_data, ["month", "sales"])

print("✅ サンプルデータを作成しました")
print("  - df_sales: 売上データ")
print("  - df_employees: 社員データ")
print("  - df_monthly: 月別売上データ")

# テンポラリビューを作成してSQLでクエリ実行

# 1. DataFrameにビュー名「sales」を付ける
df_sales.createOrReplaceTempView("sales")

print("テンポラリビュー 'sales' を作成しました")

# 2. SQLでクエリ実行
result = spark.sql("""
    SELECT * FROM sales
""")

print("\nSQLクエリの結果:")
result.show()

# 複数のDataFrameにビューを作成

# 売上データにビュー名を付ける
df_sales.createOrReplaceTempView("sales")

# 社員データにビュー名を付ける
df_employees.createOrReplaceTempView("employees")

print("テンポラリビューを作成しました:")
print("- sales（売上データ）")
print("- employees（社員データ）")

# 基本的なSELECT（社員データを使用）

# 1. 全カラム選択
print("【1】全カラム選択:")
spark.sql("SELECT * FROM employees").show()

# 2. 特定カラムのみ選択
print("【2】特定カラムのみ:")
spark.sql("""
    SELECT name, salary 
    FROM employees
""").show()

# 3. カラムに別名を付ける
print("【3】カラムに別名:")
spark.sql("""
    SELECT 
        name AS 氏名,
        salary AS 給与
    FROM employees
""").show()

# WHERE句でフィルタリング

# 1. 単純な条件
print("【1】給与が35万円以上:")
spark.sql("""
    SELECT * FROM employees
    WHERE salary >= 350000
""").show()

# 2. 複数条件（AND）
print("【2】営業部かつ給与32万円以上:")
spark.sql("""
    SELECT * FROM employees
    WHERE department = '営業部' 
      AND salary >= 320000
""").show()

# 3. IN句（複数の値のいずれかに一致）
print("【3】営業部または技術部:")
spark.sql("""
    SELECT * FROM employees
    WHERE department IN ('営業部', '技術部')
""").show()

# ORDER BYで並び替え

# 1. 降順（DESC）
print("【1】給与の降順:")
spark.sql("""
    SELECT * FROM employees
    ORDER BY salary DESC
""").show()

# 2. LIMITで件数制限（TOP3）
print("【2】給与TOP3:")
spark.sql("""
    SELECT * FROM employees
    ORDER BY salary DESC
    LIMIT 3
""").show()

# GROUP BYで集計

# 1. 部門別の集計
print("【1】部門別の人数と平均給与:")
spark.sql("""
    SELECT 
        department,
        COUNT(*) AS 人数,
        AVG(salary) AS 平均給与,
        MAX(salary) AS 最高給与,
        MIN(salary) AS 最低給与
    FROM employees
    GROUP BY department
    ORDER BY 平均給与 DESC
""").show()

# 2. HAVINGで集計後のフィルター
print("【2】平均給与が35万円以上の部門:")
spark.sql("""
    SELECT 
        department,
        AVG(salary) AS 平均給与
    FROM employees
    GROUP BY department
    HAVING AVG(salary) >= 350000
""").show()

# SQLの結果をDataFrameで更に処理

# SQLで集計
result_sql = spark.sql("""
    SELECT 
        department,
        SUM(salary) AS total_salary
    FROM employees
    GROUP BY department
""")

print("【1】SQLの結果:")
result_sql.show()

# DataFrameのメソッドで更に処理（百万円単位に変換）
result_processed = result_sql.withColumn("total_millions",
    F.round(F.col("total_salary") / 10000, 1)
)

print("【2】DataFrameで更に処理:")
result_processed.show()

# 同じ処理をDataFrameとSQLで比較

# ===== DataFrameで処理 =====
print("【DataFrame】部門別の平均給与（35万以上）:")
result_df = df_employees \
    .groupBy("department") \
    .agg(F.avg("salary").alias("avg_salary")) \
    .filter(F.col("avg_salary") >= 350000)
result_df.show()

# ===== SQLで処理 =====
print("【SQL】部門別の平均給与（35万以上）:")
result_sql = spark.sql("""
    SELECT department, AVG(salary) AS avg_salary
    FROM employees
    GROUP BY department
    HAVING AVG(salary) >= 350000
""")
result_sql.show()

print("→ 結果は同じ！書き方が違うだけ")

# ランキングをSQLで計算

print("給与ランキング（TOP3）:")
spark.sql("""
    SELECT 
        name,
        salary,
        rank
    FROM (
        SELECT 
            name,
            salary,
            ROW_NUMBER() OVER (ORDER BY salary DESC) AS rank
        FROM employees
    ) ranked
    WHERE rank <= 3
""").show()

# 前月比をSQLで計算

# 月別売上データのビューを作成
df_monthly.createOrReplaceTempView("monthly_sales")

print("前月比:")
spark.sql("""
    SELECT 
        month,
        sales,
        LAG(sales, 1) OVER (ORDER BY month) AS prev_month,
        sales - LAG(sales, 1) OVER (ORDER BY month) AS diff,
        ROUND(((sales - LAG(sales, 1) OVER (ORDER BY month)) / 
               LAG(sales, 1) OVER (ORDER BY month)) * 100, 1) AS growth_rate
    FROM monthly_sales
    ORDER BY month
""").show()

# ピボット集計をSQLで実現

print("カテゴリ別売上（ピボット形式）:")
spark.sql("""
    SELECT 
        SUM(CASE WHEN category = '食品' THEN price * quantity ELSE 0 END) AS 食品,
        SUM(CASE WHEN category = '電化製品' THEN price * quantity ELSE 0 END) AS 電化製品,
        SUM(CASE WHEN category = '衣料品' THEN price * quantity ELSE 0 END) AS 衣料品,
        SUM(price * quantity) AS 合計
    FROM sales
""").show()