# バッチ処理（これまで）
df = spark.read.csv("path/to/data")

# ストリーミング処理（今回）
df = spark.readStream.csv("path/to/data")

from pyspark.sql import SparkSession
from pyspark.sql.types import StructType, StringType, IntegerType, TimestampType

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

# スキーマを定義（ストリーミングでは必須）
schema = StructType() \
    .add("user_id", StringType()) \
    .add("action", StringType()) \
    .add("timestamp", TimestampType())

# ストリーミングでCSVを読み込み
streaming_df = spark.readStream \
    .schema(schema) \
    .csv("input_directory/")

# 処理（バッチと同じ書き方）
result = streaming_df.groupBy("action").count()

# 結果を出力
query = result.writeStream \
    .outputMode("complete") \
    .format("console") \
    .start()

# 停止するまで待機
query.awaitTermination()

# append: 新しいデータのみ出力（集計なしの場合）
query = df.writeStream.outputMode("append").format("console").start()

# complete: 全結果を出力（集計ありの場合）
query = df.writeStream.outputMode("complete").format("console").start()

# update: 更新された行のみ出力
query = df.writeStream.outputMode("update").format("console").start()

from pyspark.sql import functions as F

# スキーマ定義
schema = StructType() \
    .add("user_id", StringType()) \
    .add("action", StringType()) \
    .add("event_time", TimestampType())

# ストリーミング読み込み
events = spark.readStream \
    .schema(schema) \
    .json("events/")

# 5分間の固定ウィンドウで集計
windowed_counts = events \
    .groupBy(
        F.window("event_time", "5 minutes"),
        "action"
    ) \
    .count()

# 出力
query = windowed_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .start()

# 5分間のウィンドウを1分ごとにスライド
windowed_counts = events \
    .groupBy(
        F.window("event_time", "5 minutes", "1 minute"),
        "action"
    ) \
    .count()

# window関数の引数：
# - 第1引数: イベント時刻カラム
# - 第2引数: ウィンドウサイズ
# - 第3引数: スライド間隔（省略時は固定ウィンドウ）

# Parquetファイルに出力
query = result.writeStream \
    .outputMode("append") \
    .format("parquet") \
    .option("path", "output/streaming_results/") \
    .option("checkpointLocation", "checkpoint/") \
    .start()

# checkpointLocationは必須（障害復旧用）

# メモリ上のテーブルに出力
query = result.writeStream \
    .outputMode("complete") \
    .format("memory") \
    .queryName("realtime_counts") \
    .start()

# SQLでクエリ可能
spark.sql("SELECT * FROM realtime_counts").show()

# デフォルト：できるだけ速く処理
query = result.writeStream \
    .trigger(processingTime="0 seconds") \
    .start()

# 10秒ごとに処理
query = result.writeStream \
    .trigger(processingTime="10 seconds") \
    .start()

# 1回だけ処理（バッチ的な使い方）
query = result.writeStream \
    .trigger(once=True) \
    .start()

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

spark = SparkSession.builder \
    .appName("Realtime Access Log") \
    .getOrCreate()

# アクセスログのスキーマ
log_schema = StructType([
    StructField("timestamp", TimestampType(), True),
    StructField("user_id", StringType(), True),
    StructField("page", StringType(), True),
    StructField("action", StringType(), True)
])

# ストリーミング読み込み（新しいJSONファイルを監視）
access_logs = spark.readStream \
    .schema(log_schema) \
    .json("logs/access/")

# 5分間のウィンドウでページ別アクセス数を集計
page_counts = access_logs \
    .groupBy(
        F.window("timestamp", "5 minutes"),
        "page"
    ) \
    .agg(
        F.count("*").alias("access_count"),
        F.countDistinct("user_id").alias("unique_users")
    )

# 結果をコンソールに出力
query = page_counts.writeStream \
    .outputMode("complete") \
    .format("console") \
    .option("truncate", "false") \
    .trigger(processingTime="10 seconds") \
    .start()

query.awaitTermination()

# クエリの停止
query.stop()

# 全てのストリーミングクエリを停止
for q in spark.streams.active:
    q.stop()