DE2 — Assignment 2 : Texte — Index Inversé¶
Author : Badr TAJINI - Data Engineering II - ESIEE 2025-2026
Students : DIALLO Samba & DIOP Mouhamed
Piste : (B Archive Github)
Noms : (DIALLO Samba - DIOP Mouhamed)
Complétez les cellules ci-dessous. Consultez DE2_Lab2_Overview_EN.md et helper_assignment2-de2_esiee.md pour plus de détails.
0. Configuration initiale et préparation¶
Cette première cellule initialise l'environnement Spark et crée les répertoires nécessaires pour stocker les résultats. Nous configurons la session Spark avec les paramètres réseau appropriés et affichons des informations sur l'environnement d'exécution.
In [1]:
import os, sys, shutil, time, pathlib, csv, io
from pyspark.sql import SparkSession, functions as F
from pyspark.sql.types import StructType, StructField, StringType, LongType, ArrayType
# Configuration Spark simplifiée
spark = SparkSession.builder.appName("de2-assignment2").getOrCreate()
print(f"Spark {spark.version} | UI: http://localhost:4040")
# Création des répertoires
output_dir = pathlib.Path("outputs/lab2")
shutil.rmtree(output_dir, ignore_errors=True)
parquet_dir, csv_dir, proof_dir = output_dir / "inverted_index", output_dir / "inverted_index_csv", pathlib.Path("proof")
for d in [parquet_dir, csv_dir, proof_dir]: d.mkdir(parents=True, exist_ok=True)
print("Répertoires créés.")
WARNING: Using incubator modules: jdk.incubator.vector
Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties 26/05/14 15:23:53 WARN Utils: Your hostname, sable-ThinkPad-X1-Yoga-3rd, resolves to a loopback address: 127.0.1.1; using 10.192.33.105 instead (on interface wlp2s0) 26/05/14 15:23:53 WARN Utils: Set SPARK_LOCAL_IP if you need to bind to another address
Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties Setting default log level to "WARN". To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
26/05/14 15:23:55 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark 4.0.0 | UI: http://localhost:4040 Répertoires créés.
1. Ingestion du corpus de texte¶
Dans cette section, nous chargeons un corpus de texte réel à partir du GitHub Archive. Chaque événement GitHub devient un document avec un identifiant unique (doc_id = event ID) et un contenu textuel composé du type d'événement, du nom du repository et du login de l'acteur. Nous affichons aussi le nombre total de documents et quelques exemples pour valider le chargement du corpus.
Source de données : sample_archive_github.json - Événements GitHub publics réels
In [2]:
# === Chargement GitHub Archive et transformation en corpus textuel ===
from pyspark.sql.types import BooleanType
# Schéma GitHub Archive simplifié
schema = StructType([
StructField("id", StringType(), False),
StructField("type", StringType(), False),
StructField("repo", StructType([StructField("name", StringType(), False)]), False),
StructField("actor", StructType([StructField("login", StringType(), False)]), False),
])
# Chargement et transformation en une seule étape
archive_file = "../../sample_archive_github.json"
df_corpus = (spark.read.schema(schema).json(archive_file)
.select(
F.col("id").alias("doc_id"),
F.concat_ws(" ", F.col("type"), F.col("repo.name"), F.col("actor.login")).alias("content")
))
print(f"Corpus: {df_corpus.count()} documents GitHub | Source: {archive_file}")
df_corpus.show(10, truncate=False)
[Stage 0:> (0 + 1) / 1]
Corpus: 1000 documents GitHub | Source: ../../sample_archive_github.json
+-----------+--------------------------------------------------------------------------------------------+ |doc_id |content | +-----------+--------------------------------------------------------------------------------------------+ |45193146633|WatchEvent slashback100/presence_simulation weltenwandler | |45193146634|CreateEvent chuksdozie/dcc-webapp chuksdozie | |45193146635|PushEvent frdpzk3/ppub frdpzk3 | |45193146638|PushEvent Sandhj/ST Sandhj | |45193146652|IssuesEvent Abhay-hack/Lumina LoneWolf4713 | |45193146653|PushEvent Jackcoldtd2/Phantom-Wallet-Geteway-Solona-Api-Sdk-Connect-Web3 github-actions[bot]| |45193146662|WatchEvent yuaotian/go-cursor-help parnexcodes | |45193146665|PushEvent husnapupita/WedusKripto21 github-actions[bot] | |45193146669|CreateEvent Rahma118/Counter Rahma118 | |45193146674|PushEvent jamesreprise/infrastructure jamesreprise | +-----------+--------------------------------------------------------------------------------------------+ only showing top 10 rows
2. Normalisation du texte¶
Cette étape transforme le texte brut en tokens normalisés. Nous convertissons tout en minuscules, supprimons la ponctuation, tokenisons en mots individuels, et filtrons les mots vides (stop-words) courants qui n'apportent pas de sens sémantique. Nous affichons les comptages avant et après normalisation pour évaluer l'impact du filtrage.
In [3]:
# === Normalisation du texte : pipeline de transformation ===
# Stop-words anglais courants a filtrer (mots sans valeur semantique)
stop_words = {
"the", "a", "an", "and", "or", "but", "in", "on", "at", "to", "for",
"is", "are", "was", "were", "be", "been", "being", "have", "has", "had",
"do", "does", "did", "will", "would", "could", "should", "may", "might",
"can", "of", "with", "by", "from", "as", "it", "this", "that", "which",
"who", "what", "where", "when", "why", "how"
}
# Pipeline de normalisation en 3 etapes
# 1. Minuscules + suppression ponctuation (regex: garder seulement alphanumerique et espaces)
df_normalized = df_corpus.withColumn(
"content_clean",
F.lower(F.regexp_replace(F.col("content"), r"[^a-zA-Z0-9\s]", ""))
)
# 2. Tokenisation: split par espaces pour obtenir mots individuels
df_tokens = df_normalized.withColumn(
"tokens",
F.split(F.col("content_clean"), r"\s+")
).drop("content_clean")
# Comptage avant filtrage pour statistiques
total_tokens_before = df_tokens.select(
F.size(F.col("tokens")).alias("token_count")
).agg(F.sum("token_count")).collect()[0][0]
print(f"Tokens AVANT filtrage: {total_tokens_before}")
# 3. Explosion (1 ligne par token) + filtrage stop-words et tokens vides
df_filtered = (df_tokens
.withColumn("token", F.explode(F.col("tokens")))
.drop("tokens", "content")
.filter((F.col("token") != "") & (~F.col("token").isin(stop_words))))
# Statistiques finales
total_tokens_after = df_filtered.count()
print(f"Tokens APRES filtrage: {total_tokens_after}")
print(f"Stop-words supprimes: {total_tokens_before - total_tokens_after}")
print("\nEchantillon tokens normalises:")
df_filtered.show(10)
[Stage 4:> (0 + 1) / 1]
Tokens AVANT filtrage: 3000
Tokens APRES filtrage: 3000 Stop-words supprimes: 0 Echantillon tokens normalises: +-----------+--------------------+ | doc_id| token| +-----------+--------------------+ |45193146633| watchevent| |45193146633|slashback100prese...| |45193146633| weltenwandler| |45193146634| createevent| |45193146634| chuksdoziedccwebapp| |45193146634| chuksdozie| |45193146635| pushevent| |45193146635| frdpzk3ppub| |45193146635| frdpzk3| |45193146638| pushevent| +-----------+--------------------+ only showing top 10 rows
3. Construction de l'index inversé¶
L'index inversé est la structure clé pour les recherches textuelles rapides. Pour chaque token (terme) unique, nous regroupons tous les IDs de documents où ce terme apparaît et comptabilisons sa fréquence. Cela crée une structure permettant une recherche O(1) au lieu de scanner tous les documents.
In [4]:
# === Construction de l'index inverse: token -> [doc_ids] + frequence ===
# Aggregation par token:
# - collect_list: rassemble tous les doc_id contenant ce token
# - count: nombre total d'occurrences du token (frequence)
# - orderBy desc: termes les plus frequents en premier
df_inverted_index = (df_filtered
.groupBy("token")
.agg(
F.collect_list("doc_id").alias("doc_ids"),
F.count("*").alias("freq")
)
.orderBy(F.desc("freq")))
# Statistiques de l'index
unique_terms = df_inverted_index.count()
print(f"Termes uniques dans l'index: {unique_terms}")
print("\nTop 15 termes par frequence:")
# truncate=80 : la colonne doc_ids est un collect_list (parfois des centaines
# d'IDs) ; sans troncature une seule ligne fait >9000 caracteres et casse
# l'affichage de la page.
df_inverted_index.show(15, truncate=80)
Termes uniques dans l'index: 1398 Top 15 termes par frequence:
+-----------------------------+--------------------------------------------------------------------------------+----+ | token| doc_ids|freq| +-----------------------------+--------------------------------------------------------------------------------+----+ | pushevent|[45193146832, 45193146846, 45193146847, 45193146861, 45193146863, 45193146864...| 714| | githubactionsbot|[45193146831, 45193146861, 45193146864, 45193146917, 45193146938, 45193146980...| 99| | createevent|[45193146839, 45193146849, 45193146851, 45193146860, 45193146894, 45193146913...| 97| | pullrequestevent|[45193146837, 45193146886, 45193146945, 45193147015, 45193147088, 45193147141...| 56| | watchevent|[45193146862, 45193146986, 45193146995, 45193147012, 45193147029, 45193147432...| 50| | issuecommentevent|[45193146935, 45193146976, 45193147309, 45193147791, 45193148176, 45193148315...| 28| | dependabotbot|[45193146976, 45193147015, 45193147370, 45193147592, 45193147656, 45193147764...| 26| | aergoioherapy|[45193148410, 45193148531, 45193148698, 45193148854, 45193148966, 45193149043...| 19| | hanlsin|[45193148410, 45193148531, 45193148698, 45193148854, 45193148966, 45193149043...| 19| | releaseevent|[45193147950, 45193149627, 45193149629, 45193149635, 45193149639, 45193149653...| 14| |nilcsitopicinterviewquestions|[45193146976, 45193147015, 45193147764, 45193148912, 45193148914, 45193149031...| 13| | pullbot|[45193147322, 45193147449, 45193147859, 45193147860, 45193147964, 45193148247...| 13| | dandapandabytesdevcrate|[45193149627, 45193149629, 45193149635, 45193149639, 45193149653, 45193149656...| 13| | vpnsuperappfast|[45193147536, 45193147667, 45193147996, 45193148299, 45193148518, 45193148637...| 12| | vpnsuperapp|[45193147536, 45193147667, 45193147996, 45193148299, 45193148518, 45193148637...| 12| +-----------------------------+--------------------------------------------------------------------------------+----+ only showing top 15 rows
4. Écriture en Parquet et CSV¶
Nous persistons l'index inversé dans deux formats pour comparer leur efficacité. Le format Parquet est columnar et optimisé pour les requêtes analytiques (avec compression), tandis que CSV est un format texte universel mais moins compressé et moins rapide pour les requêtes.
In [5]:
# === Persistence de l'index: Parquet (columnar) vs CSV (texte) ===
# Ecriture Parquet: format columnar compresse, optimal pour requetes analytiques
print("Ecriture Parquet...")
df_inverted_index.write.mode("overwrite").parquet(str(parquet_dir))
print(f"Parquet: {parquet_dir}")
# Ecriture CSV: format texte universel mais moins efficace
# Transformation necessaire: array -> string separee par virgules
print("\nEcriture CSV...")
df_inverted_index_csv = df_inverted_index.withColumn(
"doc_ids",
F.concat_ws(",", F.col("doc_ids"))
)
# coalesce(1): un seul fichier CSV pour simplicite
df_inverted_index_csv.coalesce(1).write \
.mode("overwrite") \
.option("header", "true") \
.csv(str(csv_dir))
print(f"CSV: {csv_dir}")
# Rechargement pour tests de requetes
df_index_parquet = spark.read.parquet(str(parquet_dir))
df_index_csv = spark.read.option("header", "true").csv(str(csv_dir))
print("\nIndex recharges pour benchmarking.")
Ecriture Parquet...
[Stage 27:> (0 + 1) / 1]
Parquet: outputs/lab2/inverted_index Ecriture CSV...
CSV: outputs/lab2/inverted_index_csv
Index recharges pour benchmarking.
5. Mesure de la latence des requêtes¶
Nous mesurons le temps d'exécution (wall-clock time) pour rechercher des termes spécifiques dans l'index. Ceci simule un cas d'utilisation réel où les utilisateurs interrogent l'index inversé. Nous enregistrerons les plans Spark pour comprendre les optimisations appliquées par le moteur de requête.
In [6]:
# === Benchmark: mesure de latence des requetes (Parquet vs CSV) ===
# Termes a rechercher dans l'index
terms_to_query = ["pushevent", "github", "bot"]
query_latencies = []
# Test 1: Requetes sur index Parquet (format columnar compresse)
print("Benchmark Parquet:")
print("-" * 70)
for term in terms_to_query:
start_time = time.time()
result = df_index_parquet.filter(F.col("token") == term).collect()
latency_ms = (time.time() - start_time) * 1000
# Stockage des metriques pour le rapport CSV
query_latencies.append({
"term": term,
"format": "parquet",
"latency_ms": latency_ms,
"found": len(result) > 0,
"freq": result[0]["freq"] if result else 0,
"doc_count": len(result[0]["doc_ids"]) if result else 0
})
print(f"'{term}': {latency_ms:.2f} ms", end="")
if result:
print(f" | freq={result[0]['freq']}, docs={len(result[0]['doc_ids'])}")
else:
print(" | non trouve")
# Test 2: Requetes sur index CSV (format texte)
print("\nBenchmark CSV:")
print("-" * 70)
for term in terms_to_query:
start_time = time.time()
result = df_index_csv.filter(F.col("token") == term).collect()
latency_ms = (time.time() - start_time) * 1000
print(f"'{term}': {latency_ms:.2f} ms", end="")
if result:
doc_count = len(result[0]["doc_ids"].split(",")) if result[0]["doc_ids"] else 0
print(f" | freq={result[0]['freq']}, docs={doc_count}")
else:
print(" | non trouve")
print("\nLatences enregistrees.")
Benchmark Parquet: ----------------------------------------------------------------------
'pushevent': 307.43 ms | freq=714, docs=714 'github': 82.24 ms | non trouve 'bot': 69.18 ms | non trouve Benchmark CSV: ----------------------------------------------------------------------
'pushevent': 144.27 ms | freq=714, docs=714 'github': 80.52 ms | non trouve 'bot': 103.95 ms | non trouve Latences enregistrees.
6. Comparaison de l'empreinte disque¶
Nous comparons la taille disque utilisée par les deux formats de stockage. Parquet utilise la compression columnar, tandis que CSV est un format texte universel mais moins compressé et moins rapide pour les requêtes. Cette comparaison montre les économies de stockage réalisées avec des formats spécialisés.
In [7]:
# Fonction utilitaire: calcul taille totale d'un repertoire
def get_directory_size(path):
return sum(os.path.getsize(os.path.join(dp, f))
for dp, dn, filenames in os.walk(path)
for f in filenames if os.path.exists(os.path.join(dp, f)))
# Calcul des tailles sur disque
parquet_size = get_directory_size(str(parquet_dir))
csv_size = get_directory_size(str(csv_dir))
print("Empreinte disque:")
print("-" * 70)
print(f"Parquet: {parquet_size:,} bytes ({parquet_size / 1024:.1f} KB)")
print(f"CSV: {csv_size:,} bytes ({csv_size / 1024:.1f} KB)")
# Ratio de compression (negatif = Parquet plus gros, positif = Parquet plus petit)
if csv_size > 0:
ratio = (csv_size - parquet_size) / csv_size * 100
print(f"Difference: {ratio:+.1f}% (Parquet vs CSV)")
# Detail des fichiers Parquet (metadata + data + checksums)
print("\nFichiers Parquet:")
for root, dirs, files in os.walk(str(parquet_dir)):
for file in files:
size = os.path.getsize(os.path.join(root, file))
print(f" {file}: {size:,} bytes")
Empreinte disque: ---------------------------------------------------------------------- Parquet: 31,162 bytes (30.4 KB) CSV: 65,175 bytes (63.6 KB) Difference: +52.2% (Parquet vs CSV) Fichiers Parquet: ._SUCCESS.crc: 8 bytes part-00000-45372f4d-7988-4b46-aba5-3929dfc51701-c000.snappy.parquet: 30,902 bytes .part-00000-45372f4d-7988-4b46-aba5-3929dfc51701-c000.snappy.parquet.crc: 252 bytes _SUCCESS: 0 bytes
7. Preuves et métriques¶
Nous sauvegardons les plans d'exécution Spark (version textuelle et formatée) dans le répertoire proof/. Ceux-ci montrent comment Spark optimise les requêtes et où se situent les goulots d'étranglement. Nous créons également un journal de métriques CSV contenant les mesures clés : latences, tailles de stockage, et paramètres de requête.
In [8]:
# === Sauvegarde preuves: plans d'execution + metriques CSV ===
# Plan 1: Construction de l'index inverse (groupBy + aggregations)
print("Sauvegarde plan construction index...")
plan_file = proof_dir / "plan_index_build.txt"
with open(plan_file, "w") as f:
old_stdout = sys.stdout
sys.stdout = io.StringIO()
df_inverted_index.explain("formatted")
f.write(sys.stdout.getvalue())
sys.stdout = old_stdout
print(f"Plan index: {plan_file}")
# Plan 2: Requete de recherche (filter sur token)
print("Sauvegarde plan requete...")
query_plan_file = proof_dir / "plan_query.txt"
with open(query_plan_file, "w") as f:
old_stdout = sys.stdout
sys.stdout = io.StringIO()
df_index_parquet.filter(F.col("token") == "pushevent").explain("formatted")
f.write(sys.stdout.getvalue())
sys.stdout = old_stdout
print(f"Plan requete: {query_plan_file}")
# Journal des metriques: latences + tailles + statistiques
print("\nCreation journal metriques...")
metrics_log_file = "lab2_metrics_log.csv"
with open(metrics_log_file, "w", newline="") as csvfile:
writer = csv.DictWriter(csvfile, fieldnames=[
"timestamp", "index_type", "term", "latency_ms",
"parquet_size_bytes", "csv_size_bytes", "unique_terms",
"doc_count", "token_freq"
])
writer.writeheader()
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
# Ecriture des metriques de chaque requete testee
for entry in query_latencies:
writer.writerow({
"timestamp": timestamp,
"index_type": "Parquet",
"term": entry["term"],
"latency_ms": f"{entry['latency_ms']:.2f}",
"parquet_size_bytes": parquet_size,
"csv_size_bytes": csv_size,
"unique_terms": unique_terms,
"doc_count": entry["doc_count"],
"token_freq": entry["freq"]
})
print(f"Metriques: {metrics_log_file}")
print("\nCapture Spark UI: http://localhost:4040")
Sauvegarde plan construction index... Plan index: proof/plan_index_build.txt Sauvegarde plan requete... Plan requete: proof/plan_query.txt Creation journal metriques... Metriques: lab2_metrics_log.csv Capture Spark UI: http://localhost:4040
8. Nettoyage¶
Cette section finalise l'exécution en arrêtant la session Spark et en affichant un résumé des fichiers générés.
In [ ]:
spark.stop()