# SparkSessionの初期化

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

# SparkSessionを作成
spark = SparkSession.builder \
    .appName("Lazy Evaluation") \
    .getOrCreate()

print("SparkSession準備完了")

# サンプルデータを作成（10万件）

data = [(i, f"name_{i}", i % 100, i * 10) for i in range(100000)]
df = spark.createDataFrame(data, ["id", "name", "category", "value"])

print("サンプルデータ（最初の5件）:")
df.show(5)

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

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

spark = SparkSession.builder.appName("Lazy Evaluation").getOrCreate()

# サンプルデータ（10万件）
data = [(i, f"name_{i}", i % 100, i * 10) for i in range(100000)]
df = spark.createDataFrame(data, ["id", "name", "category", "value"])

print("✅ サンプルデータを作成しました")
print("  - df: 10万件のデータ（id, name, category, value）")

# Transformationは即座に実行されない

print("【Transformationのみ（実行されない）】")
start = time.time()

# これらは全てTransformation → 実行されない
df_filtered = df.filter(F.col("value") > 500000)
df_selected = df_filtered.select("id", "name")
df_sorted = df_selected.orderBy("id")

end = time.time()
print(f"処理時間: {end - start:.4f}秒")
print("→ ほぼ0秒！実際の処理は行われていない")

# Actionを実行すると処理が走る

print("【Actionを実行（ここで初めて実行）】")
start = time.time()

# show()はAction → ここで初めて実際に処理される
df_sorted.show(5)

end = time.time()
print(f"処理時間: {end - start:.2f}秒")
print("→ 時間がかかる！実際にデータ処理が行われた")

# 複数のTransformationを組み合わせる

df_result = df \
    .filter(F.col("category") < 10) \
    .filter(F.col("value") > 50000) \
    .select("name", "value") \
    .orderBy("value", ascending=False)

# 実行計画を表示
print("【実行計画】")
df_result.explain()

# フィルター処理
df_filtered = df.filter(F.col("value") > 500000)

print("【キャッシュなし】")
start = time.time()

# 1回目
count1 = df_filtered.count()
print(f"1回目 count: {count1}")

# 2回目（また最初から計算）
count2 = df_filtered.count()
print(f"2回目 count: {count2}")

end = time.time()
print(f"総処理時間: {end - start:.2f}秒")

print("【キャッシュあり】")

# キャッシュを有効化
df_filtered_cached = df_filtered.cache()

start = time.time()

# 1回目（計算 + キャッシュに保存）
count1 = df_filtered_cached.count()
print(f"1回目 count: {count1}")

# 2回目（キャッシュから読む、速い！）
count2 = df_filtered_cached.count()
print(f"2回目 count: {count2}")

end = time.time()
print(f"総処理時間: {end - start:.2f}秒")

# キャッシュ解放
df_filtered_cached.unpersist()