DE2 — Lab 2 : Traitement de texte — Pipeline d'index inversé (15%)¶
Author : Badr TAJINI - Data Engineering II - ESIEE 2025-2026
Students : DIALLO Samba & DIOP Mouhamed
Piste : Track B - GitHub Archive
Objectif : Ingérer un corpus de texte (GitHub Archive), tokeniser et normaliser, construire un index inversé, mesurer la latence des requêtes, comparer le stockage en Parquet vs CSV.
In [1]:
# === Configuration initiale ===
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, BooleanType
# Initialisation Spark simplifiee
spark = SparkSession.builder.appName("DE2-Lab2-Practice").getOrCreate()
print(f"Spark {spark.version} | UI: http://localhost:4040")
# Creation repertoires
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("Repertoires crees.")
WARNING: Using incubator modules: jdk.incubator.vector Using Spark's default log4j profile: org/apache/spark/log4j2-defaults.properties 26/05/14 13:53:33 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 13:53:33 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 13:53:35 WARN NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Spark 4.0.1 | UI: http://localhost:4040 Repertoires crees.
1. Ingestion du corpus de texte¶
Cette section charge un corpus de texte reel a partir du GitHub Archive (Track B). Chaque evenement GitHub devient un document avec un identifiant unique (doc_id = event ID) et un contenu textuel compose du type d'evenement, du nom du repository et du login de l'acteur.
Source: sample_archive_github.json - Evenements GitHub publics reels
In [2]:
# === Chargement GitHub Archive (Track B) ===
# Schema GitHub Archive simplifie
github_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: event -> document textuel
archive_file = "../../sample_archive_github.json"
df_corpus = (spark.read.schema(github_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("text")
))
# Statistiques corpus
num_docs = df_corpus.count()
avg_length = df_corpus.select(F.avg(F.length(F.col("text")))).collect()[0][0]
print(f"Corpus: {num_docs} documents GitHub | Longueur moyenne: {avg_length:.0f} caracteres")
print("\nEchantillon:")
df_corpus.show(5, truncate=80)
Corpus: 1000 documents GitHub | Longueur moyenne: 49 caracteres Echantillon: +-----------+---------------------------------------------------------+ | doc_id| text| +-----------+---------------------------------------------------------+ |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| +-----------+---------------------------------------------------------+ only showing top 5 rows
2. Normalisation du texte (Anglais) — Minuscules, tokenisation, suppression des stop-words¶
Cette etape transforme le texte brut en tokens normalises. Nous convertissons tout en minuscules, supprimons la ponctuation, tokenisons en mots individuels, et filtrons les mots vides (stop-words) courants en anglais.
In [4]:
# === Normalisation: minuscules + tokenisation + filtrage stop-words ===
# Stop-words anglais courants
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: minuscules + suppression ponctuation + tokenisation + filtrage
df_normalized = df_corpus.withColumn(
"text_clean",
F.lower(F.regexp_replace(F.col("text"), r"[^a-zA-Z0-9\s]", ""))
)
df_tokens = df_normalized.withColumn(
"tokens",
F.split(F.col("text_clean"), r"\s+")
).drop("text_clean")
# Comptage avant filtrage
total_before = df_tokens.select(F.size(F.col("tokens")).alias("count")).agg(F.sum("count")).collect()[0][0]
print(f"Tokens AVANT filtrage: {total_before}")
# Explosion + filtrage stop-words
df_filtered = (df_tokens
.withColumn("token", F.explode(F.col("tokens")))
.drop("tokens", "text")
.filter((F.col("token") != "") & (~F.col("token").isin(stop_words))))
# Statistiques finales
total_after = df_filtered.count()
print(f"Tokens APRES filtrage: {total_after}")
print(f"Stop-words supprimes: {total_before - total_after} ({(total_before - total_after) / total_before * 100:.1f}%)")
print("\nEchantillon:")
df_filtered.show(10)
Tokens AVANT filtrage: 3000 Tokens APRES filtrage: 3000 Stop-words supprimes: 0 (0.0%) Echantillon: +-----------+--------------------+ | 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 inverse¶
L'index inverse est la structure cle pour les recherches textuelles efficaces. Pour chaque token unique, nous regroupons tous les IDs de documents ou ce terme apparait et comptabilisons sa frequence. Cela permet une recherche O(1) au lieu de scanner tous les documents.
In [5]:
# Construction index inverse: token -> [doc_ids] + frequence
# Aggregation: groupBy token + collect_list doc_ids + count frequence
inverted_index = (df_filtered
.groupBy("token")
.agg(
F.collect_list("doc_id").alias("doc_ids"),
F.count("*").alias("freq")
)
.orderBy(F.desc("freq")))
# Statistiques
unique_terms = inverted_index.count()
print(f"Termes uniques: {unique_terms}")
print("\nTop 15 termes par frequence:")
inverted_index.show(15, truncate=60)
Termes uniques: 1398 Top 15 termes par frequence: +-----------------------------+------------------------------------------------------------+----+ | token| doc_ids|freq| +-----------------------------+------------------------------------------------------------+----+ | pushevent|[45193146832, 45193146846, 45193146847, 45193146861, 4519...| 714| | githubactionsbot|[45193146831, 45193146861, 45193146864, 45193146917, 4519...| 99| | createevent|[45193146839, 45193146849, 45193146851, 45193146860, 4519...| 97| | pullrequestevent|[45193146837, 45193146886, 45193146945, 45193147015, 4519...| 56| | watchevent|[45193146862, 45193146986, 45193146995, 45193147012, 4519...| 50| | issuecommentevent|[45193146935, 45193146976, 45193147309, 45193147791, 4519...| 28| | dependabotbot|[45193146976, 45193147015, 45193147370, 45193147592, 4519...| 26| | aergoioherapy|[45193148410, 45193148531, 45193148698, 45193148854, 4519...| 19| | hanlsin|[45193148410, 45193148531, 45193148698, 45193148854, 4519...| 19| | releaseevent|[45193147950, 45193149627, 45193149629, 45193149635, 4519...| 14| |nilcsitopicinterviewquestions|[45193146976, 45193147015, 45193147764, 45193148912, 4519...| 13| | pullbot|[45193147322, 45193147449, 45193147859, 45193147860, 4519...| 13| | dandapandabytesdevcrate|[45193149627, 45193149629, 45193149635, 45193149639, 4519...| 13| | vpnsuperappfast|[45193147536, 45193147667, 45193147996, 45193148299, 4519...| 12| | vpnsuperapp|[45193147536, 45193147667, 45193147996, 45193148299, 4519...| 12| +-----------------------------+------------------------------------------------------------+----+ only showing top 15 rows
4. Persistence de l'index — Parquet et CSV¶
Nous persistons l'index dans deux formats pour comparer leur efficacite. Parquet est columnar et compresse (optimal pour requetes analytiques), tandis que CSV est texte universel mais moins efficace.
In [6]:
# Parquet: format columnar compresse
print("Ecriture Parquet...")
inverted_index.write.mode("overwrite").parquet(str(parquet_dir))
print(f"Parquet: {parquet_dir}")
# CSV: array -> string separee par virgules
print("\nEcriture CSV...")
inverted_index_csv = inverted_index.withColumn(
"doc_ids",
F.concat_ws(",", F.col("doc_ids"))
)
inverted_index_csv.coalesce(1).write \
.mode("overwrite") \
.option("header", "true") \
.csv(str(csv_dir))
print(f"CSV: {csv_dir}")
# Rechargement pour benchmarking
df_index_parquet = spark.read.parquet(str(parquet_dir))
df_index_csv = spark.read.option("header", "true").csv(str(csv_dir))
print("\nIndex recharges.")
Ecriture Parquet...
Parquet: outputs/lab2/inverted_index Ecriture CSV... CSV: outputs/lab2/inverted_index_csv Index recharges.
5. Benchmark: mesure de latence des requetes¶
Nous interrogeons l'index pour rechercher des termes specifiques et mesurons le temps d'execution reel (wall-clock time). Cela simule un scenario utilisateur reel et permet de comparer Parquet vs CSV.
In [7]:
# Cache index Parquet pour mesures justes
df_index_parquet.cache()
df_index_parquet.count()
print("Index Parquet cache.\n")
# Termes a rechercher (adaptes aux donnees GitHub)
query_terms = ["pushevent", "github", "bot", "code", "bug", "actions", "workflow"]
query_latencies = []
# Test Parquet
print("Benchmark Parquet:")
for term in query_terms:
start = time.time()
result = df_index_parquet.filter(F.col("token") == term).collect()
latency_ms = (time.time() - start) * 1000
print(f"'{term}': {latency_ms:.2f} ms", end="")
if result:
freq, doc_count = result[0]["freq"], len(result[0]["doc_ids"])
print(f" | freq={freq}, docs={doc_count}")
query_latencies.append({
"term": term, "format": "parquet", "latency_ms": latency_ms,
"found": True, "freq": freq, "doc_count": doc_count
})
else:
print(" | non trouve")
query_latencies.append({
"term": term, "format": "parquet", "latency_ms": latency_ms,
"found": False, "freq": 0, "doc_count": 0
})
# Test CSV
print("\nBenchmark CSV:")
print("-" * 80)
for term in query_terms:
start = time.time()
result = df_index_csv.filter(F.col("token") == term).collect()
latency_ms = (time.time() - start) * 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" | docs={doc_count}")
else:
print(" | non trouve")
# Sauvegarde plan execution
print("\nSauvegarde 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: {query_plan_file}")
Index Parquet cache. Benchmark Parquet: 'pushevent': 345.12 ms | freq=714, docs=714 'github': 133.72 ms | non trouve 'bot': 109.59 ms | non trouve 'code': 115.00 ms | non trouve 'bug': 147.64 ms | non trouve 'actions': 88.62 ms | non trouve 'workflow': 99.53 ms | non trouve Benchmark CSV: -------------------------------------------------------------------------------- 'pushevent': 518.50 ms | docs=714 'github': 1024.05 ms | non trouve 'bot': 421.89 ms | non trouve 'code': 416.45 ms | non trouve 'bug': 224.71 ms | non trouve 'actions': 201.14 ms | non trouve 'workflow': 233.33 ms | non trouve Sauvegarde plan requete... Plan: proof/plan_query.txt
6. Comparaison empreinte disque — Parquet vs CSV¶
Nous comparons la taille disque utilisee par les deux formats. Parquet utilise la compression columnar, tandis que CSV est texte universel mais moins optimise. Cette comparaison quantifie les economies de stockage.
In [8]:
# Fonction calcul taille 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 tailles
parquet_size = get_directory_size(str(parquet_dir))
csv_size = get_directory_size(str(csv_dir))
print("Empreinte disque:")
print("-" * 80)
print(f"Parquet: {parquet_size:,} bytes ({parquet_size / 1024:.1f} KB)")
print(f"CSV: {csv_size:,} bytes ({csv_size / 1024:.1f} KB)")
if csv_size > 0:
ratio = (csv_size - parquet_size) / csv_size * 100
print(f"Economie: {csv_size - parquet_size:,} bytes ({ratio:.1f}%)")
# Detail fichiers
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")
# Sauvegarde plan construction index
print("\nSauvegarde plan construction index...")
plan_build_file = proof_dir / "plan_index_build.txt"
with open(plan_build_file, "w") as f:
old_stdout = sys.stdout
sys.stdout = io.StringIO()
inverted_index.explain("formatted")
f.write(sys.stdout.getvalue())
sys.stdout = old_stdout
print(f"Plan: {plan_build_file}")
# Creation journal metriques
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", "format", "term", "latency_ms",
"parquet_size_bytes", "csv_size_bytes", "unique_terms",
"total_docs", "doc_count", "term_freq"
])
writer.writeheader()
timestamp = time.strftime("%Y-%m-%d %H:%M:%S")
for entry in query_latencies:
writer.writerow({
"timestamp": timestamp,
"format": "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,
"total_docs": num_docs,
"doc_count": entry["doc_count"] if entry["found"] else 0,
"term_freq": entry["freq"] if entry["found"] else 0
})
print(f"Metriques: {metrics_log_file}")
print("\nCapture Spark UI: http://localhost:4040")
Empreinte disque: -------------------------------------------------------------------------------- Parquet: 31,162 bytes (30.4 KB) CSV: 65,175 bytes (63.6 KB) Economie: 34,013 bytes (52.2%) Fichiers Parquet: ._SUCCESS.crc: 8 bytes .part-00000-45ee650a-a12b-4726-9971-cd703f9a1b3b-c000.snappy.parquet.crc: 252 bytes part-00000-45ee650a-a12b-4726-9971-cd703f9a1b3b-c000.snappy.parquet: 30,902 bytes _SUCCESS: 0 bytes Sauvegarde plan construction index... Plan: proof/plan_index_build.txt Creation journal metriques... Metriques: lab2_metrics_log.csv Capture Spark UI: http://localhost:4040
In [9]:
spark.stop()
print("\nLab 2 Practice termine.")
print("\nFichiers generes:")
print(f" - Index Parquet: {parquet_dir}")
print(f" - Index CSV: {csv_dir}")
print(f" - Plans execution: {proof_dir}")
print(f" - Metriques: {metrics_log_file}")
Lab 2 Practice termine. Fichiers generes: - Index Parquet: outputs/lab2/inverted_index - Index CSV: outputs/lab2/inverted_index_csv - Plans execution: proof - Metriques: lab2_metrics_log.csv