# Importation des bibliothèques nécessaires
import os, sys, time, pathlib, json, shutil, io
from datetime import datetime, timedelta
from contextlib import redirect_stdout
import random
import pandas as pd
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, IntegerType, TimestampType, DoubleType, BooleanType, LongType

# Création de la session Spark pour Structured Streaming
spark = SparkSession.builder.appName("de2-assignment1").getOrCreate()
print("Spark:", spark.version)

Spark: 4.0.1

BASE_DIR = pathlib.Path("/home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment")
OUTPUTS_DIR, PROOF_DIR = BASE_DIR / "outputs" / "lab1", BASE_DIR / "proof"

# Nettoyer et recréer les répertoires
shutil.rmtree(OUTPUTS_DIR, ignore_errors=True)
for d in [OUTPUTS_DIR, PROOF_DIR]: d.mkdir(parents=True, exist_ok=True)

# Schéma GitHub Archive avec repo et actor imbriqués
event_schema = StructType([
    StructField("id", StringType(), False),
    StructField("type", StringType(), False),
    StructField("created_at", StringType(), False),
    StructField("public", BooleanType(), False),
    StructField("repo", StructType([
        StructField("id", LongType(), False),
        StructField("name", StringType(), False),
        StructField("url", StringType(), False),
    ]), False),
    StructField("actor", StructType([
        StructField("id", LongType(), False),
        StructField("login", StringType(), False),
        StructField("display_login", StringType(), False),
    ]), False),
])

# Paramètres streaming
EVENT_TIME_COL, WINDOW_DURATION, WATERMARK_DELAY = "event_timestamp", "1 hour", "15 minutes"
archive_file = "../../../sample_archive_github.json"

print(f"Track B | Window: {WINDOW_DURATION} | Watermark: {WATERMARK_DELAY} | Archive: {archive_file}")

Track B | Window: 1 hour | Watermark: 15 minutes | Archive: ../../../sample_archive_github.json

# Load GitHub Archive sample data
df_raw = (spark.read.schema(event_schema).json(archive_file) 
    if os.path.exists(archive_file) 
    else spark.createDataFrame([
        ("1", "PushEvent", "2026-04-27T10:15:30Z", True, (1, "tensorflow/tensorflow", "https://github.com/tensorflow/tensorflow"), (1, "user1", "user1")),
        ("2", "IssuesEvent", "2026-04-27T10:20:45Z", False, (2, "kubernetes/kubernetes", "https://github.com/kubernetes/kubernetes"), (2, "user2", "user2")),
    ], schema=event_schema))
print(f"Loaded {df_raw.count()} GitHub events")

# Transform: extract nested fields and prepare for streaming
df_events = df_raw.select(
    F.col("id"), F.col("type").alias("event_type"),
    F.to_timestamp(F.col("created_at")).alias("event_timestamp"), F.col("public"),
    F.col("repo.name").alias("repo_name"), F.col("actor.login").alias("actor_login")
)
print(f"Transformed events: {df_events.count()} records | Schema:");df_events.printSchema();df_events.show(5)

Loaded 2 GitHub events
Transformed events: 2 records | Schema:
root
 |-- id: string (nullable = false)
 |-- event_type: string (nullable = false)
 |-- event_timestamp: timestamp (nullable = true)
 |-- public: boolean (nullable = false)
 |-- repo_name: string (nullable = false)
 |-- actor_login: string (nullable = false)

+---+-----------+-------------------+------+--------------------+-----------+
| id| event_type|    event_timestamp|public|           repo_name|actor_login|
+---+-----------+-------------------+------+--------------------+-----------+
|  1|  PushEvent|2026-04-27 12:15:30|  true|tensorflow/tensor...|      user1|
|  2|IssuesEvent|2026-04-27 12:20:45| false|kubernetes/kubern...|      user2|
+---+-----------+-------------------+------+--------------------+-----------+

# Version de base : Watermark + Aggregation par fenêtre
baseline_checkpoint = (OUTPUTS_DIR / "checkpoint_baseline")
baseline_checkpoint.mkdir(parents=True, exist_ok=True)

df_stream_baseline = df_events.withWatermark(EVENT_TIME_COL, WATERMARK_DELAY)

df_agg_baseline = (df_stream_baseline
    .groupBy(
        F.window(F.col(EVENT_TIME_COL), WINDOW_DURATION), 
        F.col("event_type"), 
        F.col("repo_name")
    )
    .agg(
        F.count("*").alias("event_count"),
        F.countDistinct("actor_login").alias("unique_actors"),
        F.sum(F.when(F.col("public"), 1).otherwise(0)).alias("public_events")
    )
    .select(
        F.col("window.start").alias("window_start"),
        F.col("window.end").alias("window_end"),
        F.col("event_type"), F.col("repo_name"),
        F.col("event_count"), F.col("unique_actors"), F.col("public_events")
    ))

print("Plan requête baseline:")
df_agg_baseline.explain("formatted")

Plan requête baseline:
== Physical Plan ==
AdaptiveSparkPlan (11)
+- HashAggregate (10)
   +- Exchange (9)
      +- HashAggregate (8)
         +- HashAggregate (7)
            +- Exchange (6)
               +- HashAggregate (5)
                  +- Project (4)
                     +- Project (3)
                        +- Filter (2)
                           +- Scan ExistingRDD (1)


(1) Scan ExistingRDD
Output [6]: [id#469, type#470, created_at#471, public#472, repo#473, actor#474]
Arguments: [id#469, type#470, created_at#471, public#472, repo#473, actor#474], MapPartitionsRDD[4] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)

(2) Filter
Input [6]: [id#469, type#470, created_at#471, public#472, repo#473, actor#474]
Condition : isnotnull(cast(created_at#471 as timestamp))

(3) Project
Output [5]: [type#470 AS event_type#485, cast(created_at#471 as timestamp) AS event_timestamp#486, public#472, repo#473.name AS repo_name#487, actor#474.login AS actor_login#488]
Input [6]: [id#469, type#470, created_at#471, public#472, repo#473, actor#474]

(4) Project
Output [5]: [named_struct(start, knownnullable(precisetimestampconversion(((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - CASE WHEN (((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) < 0) THEN (((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) + 3600000000) ELSE ((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) END) - 0), LongType, TimestampType)), end, knownnullable(precisetimestampconversion((((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - CASE WHEN (((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) < 0) THEN (((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) + 3600000000) ELSE ((precisetimestampconversion(event_timestamp#486, TimestampType, LongType) - 0) % 3600000000) END) - 0) + 3600000000), LongType, TimestampType))) AS window#532, event_type#485, public#472, repo_name#487, actor_login#488]
Input [5]: [event_type#485, event_timestamp#486, public#472, repo_name#487, actor_login#488]

(5) HashAggregate
Input [5]: [window#532, event_type#485, public#472, repo_name#487, actor_login#488]
Keys [4]: [window#532, event_type#485, repo_name#487, actor_login#488]
Functions [2]: [partial_count(1), partial_sum(CASE WHEN public#472 THEN 1 ELSE 0 END)]
Aggregate Attributes [2]: [count(1)#529L, sum(CASE WHEN public#472 THEN 1 ELSE 0 END)#531L]
Results [6]: [window#532, event_type#485, repo_name#487, actor_login#488, count#538L, sum#540L]

(6) Exchange
Input [6]: [window#532, event_type#485, repo_name#487, actor_login#488, count#538L, sum#540L]
Arguments: hashpartitioning(window#532, event_type#485, repo_name#487, actor_login#488, 200), ENSURE_REQUIREMENTS, [plan_id=974]

(7) HashAggregate
Input [6]: [window#532, event_type#485, repo_name#487, actor_login#488, count#538L, sum#540L]
Keys [4]: [window#532, event_type#485, repo_name#487, actor_login#488]
Functions [2]: [merge_count(1), merge_sum(CASE WHEN public#472 THEN 1 ELSE 0 END)]
Aggregate Attributes [2]: [count(1)#529L, sum(CASE WHEN public#472 THEN 1 ELSE 0 END)#531L]
Results [6]: [window#532, event_type#485, repo_name#487, actor_login#488, count#538L, sum#540L]

(8) HashAggregate
Input [6]: [window#532, event_type#485, repo_name#487, actor_login#488, count#538L, sum#540L]
Keys [3]: [window#532, event_type#485, repo_name#487]
Functions [3]: [merge_count(1), merge_sum(CASE WHEN public#472 THEN 1 ELSE 0 END), partial_count(distinct actor_login#488)]
Aggregate Attributes [3]: [count(1)#529L, sum(CASE WHEN public#472 THEN 1 ELSE 0 END)#531L, count(actor_login#488)#530L]
Results [6]: [window#532, event_type#485, repo_name#487, count#538L, sum#540L, count#543L]

(9) Exchange
Input [6]: [window#532, event_type#485, repo_name#487, count#538L, sum#540L, count#543L]
Arguments: hashpartitioning(window#532, event_type#485, repo_name#487, 200), ENSURE_REQUIREMENTS, [plan_id=978]

(10) HashAggregate
Input [6]: [window#532, event_type#485, repo_name#487, count#538L, sum#540L, count#543L]
Keys [3]: [window#532, event_type#485, repo_name#487]
Functions [3]: [count(1), sum(CASE WHEN public#472 THEN 1 ELSE 0 END), count(distinct actor_login#488)]
Aggregate Attributes [3]: [count(1)#529L, sum(CASE WHEN public#472 THEN 1 ELSE 0 END)#531L, count(actor_login#488)#530L]
Results [7]: [window#532.start AS window_start#533, window#532.end AS window_end#534, event_type#485, repo_name#487, count(1)#529L AS event_count#520L, count(actor_login#488)#530L AS unique_actors#521L, sum(CASE WHEN public#472 THEN 1 ELSE 0 END)#531L AS public_events#522L]

(11) AdaptiveSparkPlan
Output [7]: [window_start#533, window_end#534, event_type#485, repo_name#487, event_count#520L, unique_actors#521L, public_events#522L]
Arguments: isFinalPlan=false

baseline_sink, baseline_checkpoint = OUTPUTS_DIR / "stream_sink_baseline", OUTPUTS_DIR / "checkpoint_baseline"
baseline_checkpoint.mkdir(parents=True, exist_ok=True)
df_agg_baseline.coalesce(1).write.mode("overwrite").parquet(str(baseline_sink))
baseline_output = spark.read.parquet(str(baseline_sink))
print(f"Base: {baseline_output.count()} windows | Sink: {baseline_sink}")
baseline_output.orderBy("window_start").show(10, truncate=False)

Base: 2 windows | Sink: /home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/outputs/lab1/stream_sink_baseline
+-------------------+-------------------+-----------+---------------------+-----------+-------------+-------------+
|window_start       |window_end         |event_type |repo_name            |event_count|unique_actors|public_events|
+-------------------+-------------------+-----------+---------------------+-----------+-------------+-------------+
|2026-04-27 12:00:00|2026-04-27 13:00:00|PushEvent  |tensorflow/tensorflow|1          |1            |1            |
|2026-04-27 12:00:00|2026-04-27 13:00:00|IssuesEvent|kubernetes/kubernetes|1          |1            |0            |
+-------------------+-------------------+-----------+---------------------+-----------+-------------+-------------+

# Spark UI link for proof capture
from IPython.display import HTML, display
spark_ui_html = """
<div style="background-color: #f0f8ff; padding: 15px; border-radius: 8px; border-left: 4px solid #4169e1;">
    <h3 style="margin-top: 0; color: #1e3a8a;">🔍 Spark UI - Capturer les preuves</h3>
    <p><strong>Pendant l'exécution, accédez au Spark UI:</strong></p>
    <p style="font-size: 16px;">
        <a href="http://localhost:4040" target="_blank" style="background-color: #4169e1; color: white; padding: 10px 20px; text-decoration: none; border-radius: 4px; font-weight: bold;">
            ▶ Spark UI (http://localhost:4040)
        </a>
    </p>
    <p><strong>Onglets à consulter:</strong></p>
    <ul>
        <li><strong>SQL</strong>: Plans d'exécution détaillés</li>
        <li><strong>Jobs</strong>: Timeline et stades</li>
        <li><strong>Stages</strong>: Shuffle, I/O, CPU temps</li>
        <li><strong>Executors</strong>: Utilisation mémoire/CPU</li>
    </ul>
</div>
"""
display(HTML(spark_ui_html))

print("BASE: Details d'exécution")
df_baseline_output = spark.read.parquet(str(baseline_sink))
print(f"Sortie: {df_baseline_output.count()} lignes")
df_baseline_output.explain("formatted")
baseline_plan_file = PROOF_DIR / "plan_baseline.txt"
with open(baseline_plan_file, "w") as f:
    f.write("PLAN DE REQUÊTE DE BASE\n" + "=" * 80 + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        df_baseline_output.explain("formatted")
    f.write(buf.getvalue())
print(f"Plan: {baseline_plan_file}")
df_baseline_output.orderBy("window_start").show(10)

BASE: Details d'exécution
Sortie: 2 lignes
== Physical Plan ==
* ColumnarToRow (2)
+- Scan parquet  (1)


(1) Scan parquet 
Output [7]: [window_start#279, window_end#280, event_type#281, repo_name#282, event_count#283L, unique_actors#284L, public_events#285L]
Batched: true
Location: InMemoryFileIndex [file:/home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/outputs/lab1/stream_sink_baseline]
ReadSchema: struct<window_start:timestamp,window_end:timestamp,event_type:string,repo_name:string,event_count:bigint,unique_actors:bigint,public_events:bigint>

(2) ColumnarToRow [codegen id : 1]
Input [7]: [window_start#279, window_end#280, event_type#281, repo_name#282, event_count#283L, unique_actors#284L, public_events#285L]


Plan: /home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/proof/plan_baseline.txt
+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+
|       window_start|         window_end| event_type|           repo_name|event_count|unique_actors|public_events|
+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+
|2026-04-27 12:00:00|2026-04-27 13:00:00|  PushEvent|tensorflow/tensor...|          1|            1|            1|
|2026-04-27 12:00:00|2026-04-27 13:00:00|IssuesEvent|kubernetes/kubern...|          1|            1|            0|
+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+

# --- VERSION OPTIMISÉE: AVEC REPARTITIONNEMENT ---

# Répertoire de checkpoint pour cette version optimisée
optimized_checkpoint = OUTPUTS_DIR / "checkpoint_optimized"
optimized_checkpoint.mkdir(parents=True, exist_ok=True)

# Utiliser les mêmes données GitHub (df_events) que la version de base
df_stream_opt = (df_events
    .withWatermark(EVENT_TIME_COL, WATERMARK_DELAY))

# OPTIMISATION: Repartitionner par event_type + repo_name AVANT l'agrégation fenêtre
# Cela colocalise les données d'un même type d'événement et repository sur les mêmes partitions
# et réduit le coût du shuffle lors du groupBy
df_stream_opt_repartitioned = df_stream_opt.repartition(4, F.col("event_type"), F.col("repo_name"))

# Agrégation par fenêtre OPTIMISÉE (sur données repartitionnées)
df_agg_opt = (df_stream_opt_repartitioned
    .groupBy(
        F.window(F.col(EVENT_TIME_COL), WINDOW_DURATION),
        F.col("event_type"),
        F.col("repo_name")
    )
    .agg(
        F.count("*").alias("event_count"),
        F.countDistinct("actor_login").alias("unique_actors"),
        F.sum(F.when(F.col("public"), 1).otherwise(0)).alias("public_events"),
    )
    .select(
        F.col("window.start").alias("window_start"),
        F.col("window.end").alias("window_end"),
        F.col("event_type"),
        F.col("repo_name"),
        F.col("event_count"),
        F.col("unique_actors"),
        F.col("public_events"),
    ))
optimized_sink = OUTPUTS_DIR / "stream_sink_optimized"
df_agg_opt.explain("formatted")
df_agg_opt.coalesce(1).write.mode("overwrite").parquet(str(optimized_sink))
print(f"Optimized sink: {optimized_sink}")

== Physical Plan ==
AdaptiveSparkPlan (10)
+- HashAggregate (9)
   +- HashAggregate (8)
      +- HashAggregate (7)
         +- HashAggregate (6)
            +- Project (5)
               +- Exchange (4)
                  +- Project (3)
                     +- Filter (2)
                        +- Scan ExistingRDD (1)


(1) Scan ExistingRDD
Output [6]: [id#152, type#153, created_at#154, public#155, repo#156, actor#157]
Arguments: [id#152, type#153, created_at#154, public#155, repo#156, actor#157], MapPartitionsRDD[4] at applySchemaToPythonRDD at NativeMethodAccessorImpl.java:0, ExistingRDD, UnknownPartitioning(0)

(2) Filter
Input [6]: [id#152, type#153, created_at#154, public#155, repo#156, actor#157]
Condition : isnotnull(cast(created_at#154 as timestamp))

(3) Project
Output [5]: [type#153 AS event_type#168, cast(created_at#154 as timestamp) AS event_timestamp#169, public#155, repo#156.name AS repo_name#170, actor#157.login AS actor_login#171]
Input [6]: [id#152, type#153, created_at#154, public#155, repo#156, actor#157]

(4) Exchange
Input [5]: [event_type#168, event_timestamp#169, public#155, repo_name#170, actor_login#171]
Arguments: hashpartitioning(event_type#168, repo_name#170, 4), REPARTITION_BY_NUM, [plan_id=514]

(5) Project
Output [5]: [named_struct(start, knownnullable(precisetimestampconversion(((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - CASE WHEN (((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) < 0) THEN (((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) + 3600000000) ELSE ((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) END) - 0), LongType, TimestampType)), end, knownnullable(precisetimestampconversion((((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - CASE WHEN (((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) < 0) THEN (((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) + 3600000000) ELSE ((precisetimestampconversion(event_timestamp#169, TimestampType, LongType) - 0) % 3600000000) END) - 0) + 3600000000), LongType, TimestampType))) AS window#332, event_type#168, public#155, repo_name#170, actor_login#171]
Input [5]: [event_type#168, event_timestamp#169, public#155, repo_name#170, actor_login#171]

(6) HashAggregate
Input [5]: [window#332, event_type#168, public#155, repo_name#170, actor_login#171]
Keys [4]: [window#332, event_type#168, repo_name#170, actor_login#171]
Functions [2]: [partial_count(1), partial_sum(CASE WHEN public#155 THEN 1 ELSE 0 END)]
Aggregate Attributes [2]: [count(1)#329L, sum(CASE WHEN public#155 THEN 1 ELSE 0 END)#331L]
Results [6]: [window#332, event_type#168, repo_name#170, actor_login#171, count#338L, sum#340L]

(7) HashAggregate
Input [6]: [window#332, event_type#168, repo_name#170, actor_login#171, count#338L, sum#340L]
Keys [4]: [window#332, event_type#168, repo_name#170, actor_login#171]
Functions [2]: [merge_count(1), merge_sum(CASE WHEN public#155 THEN 1 ELSE 0 END)]
Aggregate Attributes [2]: [count(1)#329L, sum(CASE WHEN public#155 THEN 1 ELSE 0 END)#331L]
Results [6]: [window#332, event_type#168, repo_name#170, actor_login#171, count#338L, sum#340L]

(8) HashAggregate
Input [6]: [window#332, event_type#168, repo_name#170, actor_login#171, count#338L, sum#340L]
Keys [3]: [window#332, event_type#168, repo_name#170]
Functions [3]: [merge_count(1), merge_sum(CASE WHEN public#155 THEN 1 ELSE 0 END), partial_count(distinct actor_login#171)]
Aggregate Attributes [3]: [count(1)#329L, sum(CASE WHEN public#155 THEN 1 ELSE 0 END)#331L, count(actor_login#171)#330L]
Results [6]: [window#332, event_type#168, repo_name#170, count#338L, sum#340L, count#343L]

(9) HashAggregate
Input [6]: [window#332, event_type#168, repo_name#170, count#338L, sum#340L, count#343L]
Keys [3]: [window#332, event_type#168, repo_name#170]
Functions [3]: [count(1), sum(CASE WHEN public#155 THEN 1 ELSE 0 END), count(distinct actor_login#171)]
Aggregate Attributes [3]: [count(1)#329L, sum(CASE WHEN public#155 THEN 1 ELSE 0 END)#331L, count(actor_login#171)#330L]
Results [7]: [window#332.start AS window_start#333, window#332.end AS window_end#334, event_type#168, repo_name#170, count(1)#329L AS event_count#320L, count(actor_login#171)#330L AS unique_actors#321L, sum(CASE WHEN public#155 THEN 1 ELSE 0 END)#331L AS public_events#322L]

(10) AdaptiveSparkPlan
Output [7]: [window_start#333, window_end#334, event_type#168, repo_name#170, event_count#320L, unique_actors#321L, public_events#322L]
Arguments: isFinalPlan=false


Optimized sink: /home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/outputs/lab1/stream_sink_optimized

print("OPTIMIZED: Details")
df_optimized_output = spark.read.parquet(str(optimized_sink))
df_optimized_output.explain("formatted")
optimized_plan_file = PROOF_DIR / "plan_optimized.txt"
with open(optimized_plan_file, "w") as f:
    f.write("PLAN DE REQUÊTE OPTIMISÉE (avec repartitionnement)\n" + "=" * 80 + "\n")
    buf = io.StringIO()
    with redirect_stdout(buf):
        df_optimized_output.explain("formatted")
    f.write(buf.getvalue())
df_optimized_output.orderBy("window_start").show(10)
metrics_data = {
    "version": ["base", "optimized"],
    "optimization": ["No repartitioning", "Repartition by event_type + repo_name"],
    "output_rows": [df_baseline_output.count(), df_optimized_output.count()],
    "avg_event_count": [df_baseline_output.agg(F.avg("event_count")).collect()[0][0], df_optimized_output.agg(F.avg("event_count")).collect()[0][0]],
    "avg_unique_actors": [df_baseline_output.agg(F.avg("unique_actors")).collect()[0][0], df_optimized_output.agg(F.avg("unique_actors")).collect()[0][0]],
}
df_metrics, metrics_csv = pd.DataFrame(metrics_data), BASE_DIR / "lab1_metrics_log.csv"
df_metrics.to_csv(metrics_csv, index=False)

OPTIMIZED: Details
== Physical Plan ==
* ColumnarToRow (2)
+- Scan parquet  (1)


(1) Scan parquet 
Output [7]: [window_start#349, window_end#350, event_type#351, repo_name#352, event_count#353L, unique_actors#354L, public_events#355L]
Batched: true
Location: InMemoryFileIndex [file:/home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/outputs/lab1/stream_sink_optimized]
ReadSchema: struct<window_start:timestamp,window_end:timestamp,event_type:string,repo_name:string,event_count:bigint,unique_actors:bigint,public_events:bigint>

(2) ColumnarToRow [codegen id : 1]
Input [7]: [window_start#349, window_end#350, event_type#351, repo_name#352, event_count#353L, unique_actors#354L, public_events#355L]


+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+
|       window_start|         window_end| event_type|           repo_name|event_count|unique_actors|public_events|
+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+
|2026-04-27 12:00:00|2026-04-27 13:00:00|  PushEvent|tensorflow/tensor...|          1|            1|            1|
|2026-04-27 12:00:00|2026-04-27 13:00:00|IssuesEvent|kubernetes/kubern...|          1|            1|            0|
+-------------------+-------------------+-----------+--------------------+-----------+-------------+-------------+

# Engineering note & GENAI declaration
eng_note = BASE_DIR / "ENGINEERING_NOTE.md"
with open(eng_note, "w") as f:
    f.write("""# DE2 Lab 1: Streaming Pipeline - GitHub Archive Track B

## Objective
Implement and optimize a Spark streaming pipeline on GitHub Archive (Track B) with window aggregation, watermark management, and Parquet persistence with baseline vs optimized comparison.

## Architecture
**Baseline**: No repartitioning, Parquet sink baseline_sink
**Optimized**: Repartition by event_type + repo_name (4 partitions) before aggregation

## Data: GitHub Archive (real public events)
- Window: 1 hour | Watermark: 15 minutes
- Aggregations: count(*), countDistinct(actor_login), sum(public events)
- Outputs: plan_baseline.txt, plan_optimized.txt, lab1_metrics_log.csv
""")

genai_file = BASE_DIR / "GENAI.md"
with open(genai_file, "w") as f:
    f.write("# GENAI.md\\nNo generative AI used. All code manually written and tested.\\nData source: GitHub Archive real public events.\\n")

print(f"Completed: {eng_note}, {genai_file}")
spark.stop()

Completed: /home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/ENGINEERING_NOTE.md, /home/sable/Documents/E4FD/S4/Data Engineering/Data Engineering 2/lab1 assignment/GENAI.md

DE2 -- Assignment 1: Streaming Pipeline with GitHub Archive¶

0. Setup¶

1. Schema Definition and GitHub Archive Source¶

Chargement du GitHub Archive (Track B)¶

2. Agrégation par fenêtre + watermark (Version de base)¶

2.1 Accès au Spark UI pour capture des preuves¶

3. Écriture du flux vers Parquet¶

4. Suivi et capture des preuves (Version de base)¶

5. Optimisation et re-mesure (Version optimisée)¶

6. Remplir le journal des métriques¶

7. Nettoyage¶