Refactor: Update notebook structure for sentiment analysis and streaming queries

Author: h00dieB0y

notebooks/03_silver_to_gold.ipynb

@@ -306,8 +306,21 @@
    "cell_type": "markdown",
    "metadata": {},
    "source": [
-    "## 6. SparkSQL - Sentiment par domaine\n",
-    "\n",
+    "## 6. Requêtes en batch"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "### 6a. SparkSQL - Sentiment par domaine\n",
+    "\n"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
     "Jointure comments + stories pour répondre à : *\"Quels sites génèrent les discussions les plus positives/négatives ?\"*"
    ]
   },
@@ -341,11 +354,18 @@
   },
   {
    "cell_type": "markdown",
-   "source": "## 6b. Batch avec Window Function - Classement des auteurs\n\nUtilisation de RANK() pour classer les auteurs par leur ratio de commentaires positifs.",
-   "metadata": {}
+   "metadata": {},
+   "source": [
+    "## 6b. Window Function - Classement des auteurs\n",
+    "\n",
+    "Utilisation de RANK() pour classer les auteurs par leur ratio de commentaires positifs."
+   ]
   },
   {
    "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
    "source": [
     "# Agrégation par auteur : total comments, positifs, négatifs\n",
     "authors_stats = comments_sentiment \\\n",
@@ -366,10 +386,103 @@
     "    .filter(col(\"rank\") <= 20)\n",
     "\n",
     "authors_ranked.show(20, truncate=False)\n"
-   ],
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 7. Requêtes en streaming"
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "comments_stream = spark.readStream.format(\"delta\").load(f\"{SILVER_PATH}/comments\")\n",
+    "comments_with_sentiment = sentiment_model_fitted.transform(comments_stream)                                                                                                                  "
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Agrégation du sentiment des commentaires par fenêtre de 5minutes:                                                                                                                                            "
+   ]
+  },
+  {
+   "cell_type": "code",
+   "execution_count": null,
    "metadata": {},
+   "outputs": [],
+   "source": [
+    "sentiment_windowed = comments_with_sentiment \\\n",
+    "    .withColumn(\"sentiment_result\", explode(col(\"sentiment.result\"))) \\\n",
+    "    .groupBy(\n",
+    "        window(\n",
+    "            col(\"timestamp\"),\n",
+    "            \"5 minutes\",\n",
+    "        )\n",
+    "    ) \\\n",
+    "    .agg(\n",
+    "        count(\"*\").alias(\"total_comments\"),\n",
+    "        sum_(when(col(\"sentiment_result\") == \"pos\", 1).otherwise(0)).alias(\"positive_count\"),\n",
+    "        sum_(when(col(\"sentiment_result\") == \"neg\", 1).otherwise(0)).alias(\"negative_count\")\n",
+    "    ) \\\n",
+    "    .withColumn(\"positive_ratio\", round_(col(\"positive_count\") * 100.0 / col(\"total_comments\"), 2))\n",
+    "\n",
+    "sentiment_windowed.writeStream \\\n",
+    "    .format(\"delta\") \\\n",
+    "    .outputMode(\"update\") \\\n",
+    "    .option(\"checkpointLocation\", f\"{GOLD_PATH}/_checkpoints/sentiment_windowed\") \\\n",
+    "    .start(f\"{GOLD_PATH}/sentiment_real_time\")"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "## 7a. Sentiment en temps réel"
+   ]
+  },
+  {
+   "cell_type": "markdown",
+   "metadata": {},
+   "source": [
+    "Top mots-clés avec fenêtre glissante de 10 min, mise à jour toutes les 2 min:"
+   ]
+  },
+  {
+   "cell_type": "code",
    "execution_count": null,
-   "outputs": []
+   "metadata": {},
+   "outputs": [],
+   "source": [
+    "comments_with_kw = keywords_model_fitted.transform(comments_stream)\n",
+    "keywords_exploded_stream = comments_with_kw \\\n",
+    "    .select(\n",
+    "        col(\"id\"),\n",
+    "        explode(col(\"keywords\")).alias(\"keyword\")\n",
+    "    ) \\\n",
+    "    .groupBy(\n",
+    "        window(\n",
+    "            col(\"timestamp\"),\n",
+    "            \"10 minutes\",\n",
+    "            \"2 minutes\"\n",
+    "        ),\n",
+    "        col(\"keyword\")\n",
+    "    ) \\\n",
+    "    .count() \\\n",
+    "    .orderBy(desc(\"count\"))\n",
+    "\n",
+    "keywords_exploded_stream.writeStream \\\n",
+    "    .format(\"delta\") \\\n",
+    "    .outputMode(\"complete\") \\\n",
+    "    .option(\"checkpointLocation\", f\"{GOLD_PATH}/_checkpoints/keywords_windowed\") \\\n",
+    "    .start(f\"{GOLD_PATH}/keywords_real_time\")"
+   ]
   },
   {
    "cell_type": "markdown",
@@ -531,4 +644,4 @@
  },
  "nbformat": 4,
  "nbformat_minor": 4
-}
+}