===== lab06 - Wed Nov  4 12:52:02 PM UTC 2026 =====
$ cd /home/onyxia/work/ece-2026-bigdata
$ aws s3 --profile default cp s3://user-m-anane-ece/bronze/users.csv users.csv
Completed 7.2 KiB/7.2 KiB (79.7 KiB/s) with 1 file(s) remaining
download: s3://user-m-anane-ece/bronze/users.csv to ./users.csv
$ aws s3 --profile default cp s3://user-m-anane-ece/bronze/orders.csv orders.csv
Completed 256.0 KiB/323.8 KiB (2.7 MiB/s) with 1 file(s) remaining
Completed 323.8 KiB/323.8 KiB (3.3 MiB/s) with 1 file(s) remaining
download: s3://user-m-anane-ece/bronze/orders.csv to ./orders.csv 
$ ls -la users.csv orders.csv
-rw-r--r-- 1 onyxia users 331568 Nov  4 12:47 orders.csv
-rw-r--r-- 1 onyxia users   7351 Nov  4 12:47 users.csv
$ spark-submit --master 'local[*]' --conf spark.ui.showConsoleProgress=false /home/onyxia/work/labkit/cells.py /home/onyxia/work/labkit/lab06.py

>>> [cellule 0]
... # Lab 06 - Spark RDD API (fichiers locaux users.csv / orders.csv)
... from pyspark.sql import SparkSession
... from pyspark import SparkConf, SparkContext
... import json, time, urllib.request
... spark = SparkSession.builder.appName("lab-rdd").master("local[*]").getOrCreate()
... sc = spark.sparkContext
... sc.setLogLevel("ERROR")
... def stages():
...     # Equivalent de l'onglet Stages de la Spark UI (API REST sur le port 4040)
...     app = sc.applicationId
...     data = json.load(urllib.request.urlopen(f"http://localhost:4040/api/v1/applications/{app}/stages"))
...     for s in sorted(data, key=lambda s: s["stageId"]):
...         print(s["stageId"], s["status"], s["name"][:60], "tasks=", s["numTasks"], "shuffleRead=", s["shuffleReadBytes"], "shuffleWrite=", s["shuffleWriteBytes"])
... sc.defaultParallelism
2

>>> [cellule 1]
... users_lines = sc.textFile("users.csv")
... orders_lines = sc.textFile("orders.csv")
... print(users_lines.getNumPartitions())
... print(orders_lines.getNumPartitions())
2
2

>>> [cellule 2]
... orders_repartitioned = orders_lines.repartition(8)
... print(orders_repartitioned.getNumPartitions())
... print(orders_repartitioned.coalesce(2).getNumPartitions())
... print(orders_repartitioned.coalesce(2).toDebugString().decode())
8
2
(2) CoalescedRDD[10] at coalesce at NativeMethodAccessorImpl.java:0 []
 |  MapPartitionsRDD[8] at coalesce at NativeMethodAccessorImpl.java:0 []
 |  CoalescedRDD[7] at coalesce at NativeMethodAccessorImpl.java:0 []
 |  ShuffledRDD[6] at coalesce at NativeMethodAccessorImpl.java:0 []
 +-(2) MapPartitionsRDD[5] at coalesce at NativeMethodAccessorImpl.java:0 []
    |  PythonRDD[4] at RDD at PythonRDD.scala:58 []
    |  orders.csv MapPartitionsRDD[3] at textFile at NativeMethodAccessorImpl.java:0 []
    |  orders.csv HadoopRDD[2] at textFile at NativeMethodAccessorImpl.java:0 []

>>> [cellule 3]
... header = orders_lines.first()
... orders_data = orders_lines.filter(lambda line: line != header and line.strip() != "")
... print(orders_data.toDebugString().decode())
(2) PythonRDD[12] at RDD at PythonRDD.scala:58 []
 |  orders.csv MapPartitionsRDD[3] at textFile at NativeMethodAccessorImpl.java:0 []
 |  orders.csv HadoopRDD[2] at textFile at NativeMethodAccessorImpl.java:0 []

>>> [cellule 4]
... print(orders_data.count())
... print(orders_data.count())
... stages()
2829
2829
0 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 0 shuffleWrite= 0
1 COMPLETE count at <cell>:1 tasks= 2 shuffleRead= 0 shuffleWrite= 0
2 COMPLETE count at <cell>:2 tasks= 2 shuffleRead= 0 shuffleWrite= 0

>>> [cellule 5]
... for line in users_lines.take(6):
...     print(line)
... print(users_lines.count())
uuid,username,name,sex,address,mail,birthdate
bdd640fb-0667-4ad1-9c80-317fa3b1799d,garzaanthony,Charles Garcia,M,"908 Jennifer Squares
Robinsonshire, KY 01352",helenpeterson@gmail.com,1935-10-26
17be3111-1a2a-43ed-962b-0f79c37459ee,blairamanda,Ryan Munoz,M,"Unit 6184 Box 9593
DPO AP 09617",stanleykendra@gmail.com,2010-01-21
47294739-614f-43d7-99db-3ad0ddd1dfb2,elizabethmiles,Jacob Wood,M,"283 Steven Groves
101

>>> [cellule 6]
... import re
... 
... UUID_AT_START = re.compile(r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}")
... 
... users_records = users_lines.filter(lambda line: UUID_AT_START.match(line) is not None)
... print(users_records.count())
50

>>> [cellule 7]
... def parse_order(line):
...     order_id, user_id, date, quantity, product = line.split(",")
...     return {
...         "order_id": order_id,
...         "user_id": user_id,
...         "date": date,
...         "quantity": int(quantity),
...         "product": product.strip().lower(),
...     }
... 
... orders_rdd = orders_data.map(parse_order).cache()
... orders_rdd.take(3)
[{'order_id': '2339ba19-2563-4cc3-8a97-ebf555d596af', 'user_id': 'bdd640fb-0667-4ad1-9c80-317fa3b1799d', 'date': '2020-01-01 00:02:17.891488+00:00', 'quantity': 4, 'product': 'drink'}, {'order_id': 'b49e04cc-c243-49e7-a387-eaf6d4c3a832', 'user_id': 'bdd640fb-0667-4ad1-9c80-317fa3b1799d', 'date': '2020-01-01 01:28:08.796458+00:00', 'quantity': 5, 'product': 'bread'}, {'order_id': '41843b03-04dd-4054-9448-23f7e7e646c7', 'user_id': 'bdd640fb-0667-4ad1-9c80-317fa3b1799d', 'date': '2020-01-01 02:12:56.357080+00:00', 'quantity': 2, 'product': 'donut'}]

>>> [cellule 8]
... from operator import add
... 
... user_quantity = orders_rdd.map(lambda o: (o["user_id"], o["quantity"])).reduceByKey(add)
... user_quantity.take(5)
[('f143262f-dc5c-4eed-8da0-365bf89897b9', 12), ('969b6662-0562-4059-968c-c69b1064005c', 40), ('ce9e1a11-fcbb-4e59-bbdd-cf7c9c96e9ec', 253), ('5563f616-00e8-4ece-8b49-452d46d483f3', 57), ('f94d6204-6808-493f-9fed-2c43e256a6dc', 175)]

>>> [cellule 9]
... user_quantity_grouped = (
...     orders_rdd
...     .map(lambda o: (o["user_id"], o["quantity"]))
...     .groupByKey()
...     .mapValues(sum)
... )
... user_quantity_grouped.take(5)
[('f143262f-dc5c-4eed-8da0-365bf89897b9', 12), ('969b6662-0562-4059-968c-c69b1064005c', 40), ('ce9e1a11-fcbb-4e59-bbdd-cf7c9c96e9ec', 253), ('5563f616-00e8-4ece-8b49-452d46d483f3', 57), ('f94d6204-6808-493f-9fed-2c43e256a6dc', 175)]

>>> [cellule 10]
... # Comparaison des shuffles reduceByKey / groupByKey (onglet Stages de la Spark UI)
... stages()
0 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 0 shuffleWrite= 0
1 COMPLETE count at <cell>:1 tasks= 2 shuffleRead= 0 shuffleWrite= 0
2 COMPLETE count at <cell>:2 tasks= 2 shuffleRead= 0 shuffleWrite= 0
3 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 0 shuffleWrite= 0
4 COMPLETE count at <cell>:3 tasks= 2 shuffleRead= 0 shuffleWrite= 0
5 COMPLETE count at <cell>:6 tasks= 2 shuffleRead= 0 shuffleWrite= 0
6 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 0 shuffleWrite= 0
7 COMPLETE reduceByKey at <cell>:3 tasks= 2 shuffleRead= 0 shuffleWrite= 2513
8 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 1065 shuffleWrite= 0
9 COMPLETE groupByKey at <cell>:4 tasks= 2 shuffleRead= 0 shuffleWrite= 6230
10 COMPLETE runJob at PythonRDD.scala:191 tasks= 1 shuffleRead= 2643 shuffleWrite= 0

>>> [cellule 11]
... def most_recent(order_a, order_b):
...     return order_a if order_a["date"] > order_b["date"] else order_b
... 
... orders_dedup = (
...     orders_rdd
...     .map(lambda o: (o["order_id"], o))
...     .reduceByKey(most_recent)
...     .values()
... )
... 
... print(orders_rdd.count(), orders_dedup.count())
2829 2829

>>> [cellule 12]
... def parse_user_for_join(line):
...     fields = line.split(",")
...     return fields[0], fields[1]  # uuid, username
... 
... users_kv = users_records.map(parse_user_for_join)
... orders_kv = orders_rdd.map(lambda o: (o["user_id"], o))
... 
... enriched = orders_kv.join(users_kv)
... enriched.take(3)
[('f143262f-dc5c-4eed-8da0-365bf89897b9', ({'order_id': 'f1f7d904-1307-4157-98c1-28e7d6cd3155', 'user_id': 'f143262f-dc5c-4eed-8da0-365bf89897b9', 'date': '2020-01-07 17:49:04.785279+00:00', 'quantity': 1, 'product': 'bread'}, 'kayla51')), ('f143262f-dc5c-4eed-8da0-365bf89897b9', ({'order_id': 'a175a9c3-d7b1-42a9-a332-a50e6992896e', 'user_id': 'f143262f-dc5c-4eed-8da0-365bf89897b9', 'date': '2020-01-07 18:13:53.160212+00:00', 'quantity': 2, 'product': 'bread'}, 'kayla51')), ('f143262f-dc5c-4eed-8da0-365bf89897b9', ({'order_id': '5f8a14bd-74bd-4c93-8177-b66137fe032c', 'user_id': 'f143262f-dc5c-4eed-8da0-365bf89897b9', 'date': '2020-01-07 19:03:37.620428+00:00', 'quantity': 5, 'product': 'croissant'}, 'kayla51'))]

>>> [cellule 13]
... left = orders_kv.leftOuterJoin(users_kv)
... print(left.count(), left.filter(lambda kv: kv[1][1] is None).count())
2829 0

>>> [cellule 14]
... # Exercice 1 : aggregateByKey -> (nombre de commandes, quantite totale) puis moyenne
... stats = orders_rdd.map(lambda o: (o["user_id"], o["quantity"])).aggregateByKey(
...     (0, 0),
...     lambda acc, q: (acc[0] + 1, acc[1] + q),
...     lambda a, b: (a[0] + b[0], a[1] + b[1]),
... )
... avg_quantity = stats.mapValues(lambda c: round(c[1] / c[0], 2))
... print(stats.take(3))
... print(avg_quantity.take(3))
[('f143262f-dc5c-4eed-8da0-365bf89897b9', (4, 12)), ('969b6662-0562-4059-968c-c69b1064005c', (12, 40)), ('ce9e1a11-fcbb-4e59-bbdd-cf7c9c96e9ec', (85, 253))]
[('f143262f-dc5c-4eed-8da0-365bf89897b9', 3.0), ('969b6662-0562-4059-968c-c69b1064005c', 3.33), ('ce9e1a11-fcbb-4e59-bbdd-cf7c9c96e9ec', 2.98)]

>>> [cellule 15]
... # Exercice 2 : un user_id inconnu injecte, puis leftOuterJoin
... fake = sc.parallelize([{"order_id": "fake-order", "user_id": "00000000-0000-0000-0000-000000000000", "date": "2020-01-01", "quantity": 1, "product": "bread"}])
... orders_with_fake = orders_rdd.union(fake)
... unmatched = orders_with_fake.map(lambda o: (o["user_id"], o)).leftOuterJoin(users_kv).filter(lambda kv: kv[1][1] is None)
... print("orders without user:", unmatched.count())
orders without user: 1

>>> [cellule 16]
... # Exercice 3 : la meme agregation avec l'API DataFrame
... df = spark.read.csv("orders.csv", header=True, inferSchema=True)
... df_quantity = df.groupBy("user_uuid").sum("quantity")
... df_quantity.show(5, truncate=False)
... df_quantity.explain()
... print(user_quantity.toDebugString().decode())
+------------------------------------+-------------+
|user_uuid                           |sum(quantity)|
+------------------------------------+-------------+
|d47d577b-fa5a-41ca-859d-d55d4b943e30|80           |
|0b2f6d5c-700b-4d5f-b89f-72f32a60c652|190          |
|d9e604b3-8412-4335-988c-656db61e5fdb|184          |
|97fa7f04-8363-4007-9cae-9610c72c1fe3|285          |
|2051acef-097a-4e10-b6fe-bc0e7ecddbaf|301          |
+------------------------------------+-------------+
only showing top 5 rows
== Physical Plan ==
AdaptiveSparkPlan isFinalPlan=false
+- HashAggregate(keys=[user_uuid#18], functions=[sum(quantity#20)])
   +- Exchange hashpartitioning(user_uuid#18, 200), ENSURE_REQUIREMENTS, [plan_id=76]
      +- HashAggregate(keys=[user_uuid#18], functions=[partial_sum(quantity#20)])
         +- FileScan csv [user_uuid#18,quantity#20] Batched: false, DataFilters: [], Format: CSV, Location: InMemoryFileIndex(1 paths)[file:/home/onyxia/work/ece-2026-bigdata/orders.csv], PartitionFilters: [], PushedFilters: [], ReadSchema: struct<user_uuid:string,quantity:int>


(2) PythonRDD[87] at RDD at PythonRDD.scala:58 []
 |  MapPartitionsRDD[23] at mapPartitions at PythonRDD.scala:170 []
 |  ShuffledRDD[22] at partitionBy at NativeMethodAccessorImpl.java:0 []
 +-(2) PairwiseRDD[21] at reduceByKey at <cell>:3 []
    |  PythonRDD[20] at reduceByKey at <cell>:3 []
    |  PythonRDD[18] at RDD at PythonRDD.scala:58 []
    |      CachedPartitions: 2; MemorySize: 168.0 KiB; DiskSize: 0.0 B
    |  orders.csv MapPartitionsRDD[3] at textFile at NativeMethodAccessorImpl.java:0 []
    |  orders.csv HadoopRDD[2] at textFile at NativeMethodAccessorImpl.java:0 []

>>> [cellule 17]
... # Exercice 4 : temps avec et sans cache
... t = time.perf_counter(); orders_rdd.count(); user_quantity.count(); print("avec cache  : %.3f s" % (time.perf_counter() - t))
... print("stockage:", [(r.id(), r.getStorageLevel()) for r in [orders_rdd]], sc._jsc.sc().getRDDStorageInfo().length)
... orders_nocache = orders_data.map(parse_order)
... uq_nocache = orders_nocache.map(lambda o: (o["user_id"], o["quantity"])).reduceByKey(add)
... t = time.perf_counter(); orders_nocache.count(); uq_nocache.count(); print("sans cache  : %.3f s" % (time.perf_counter() - t))
avec cache  : 0.491 s
stockage: [(18, StorageLevel(False, True, False, False, 1))] <py4j.java_gateway.JavaMember object at 0x719545389e20>
sans cache  : 0.802 s

>>> [cellule 18]
... spark.stop()
$ rm -f users.csv orders.csv
$ true
EXIT_CODE=0