02 - Traitements (AWS-EMR).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# **Réalisez un traitement dans un environnement Big Data sur le Cloud**\n",
    "\n",
    "Puisque les tests en local se sont bien déroulés, nous allons maintenant exécuter notre script dans un environnement cloud sur un plus grand nombre d'images.\n",
    "\n",
    "Nous allons passer par la plateforme [AWS](https://aws.amazon.com/fr/) et utiliser deux services : \n",
    "- [EMR](https://aws.amazon.com/fr/emr/) : notre cluster de machine qui exécutera notre notebook avec Spark, grâce à JupyterHub qui sera installé sur le driver.\n",
    "- [S3](https://aws.amazon.com/fr/s3/) : notre service de stockage qui contiendra nos images à traiter, le notebook à exécuter, les logs d'exécutions de Spark et nos fichiers de sortie."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **Démarrage de la session Spark**\n",
    "\n",
    "Le code sera exécuté depuis **JupyterHub hébergé sur un cluster EMR AWS**.\n",
    "\n",
    "<u>Avant de commencer</u>, **il faut s'assurer d'utiliser le kernel PySpark**.\n",
    "\n",
    "**En utilisant ce kernel, une session spark est créé à l'exécution de la première cellule**. Les variable *spark* et *sc* seront déjà disponibles.\n",
    "\n",
    "Il n'est donc **plus nécessaire d'exécuter le code \"spark = (SparkSession ...\"** comme dans notre notebook exécuté en local."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# L'exécution de cette cellule démarre l'application Spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<u>Affichage des informations sur la session en cours et liens vers Spark UI</u> :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "%%info"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **Import des librairies**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Pour lire les images qui auront été chargées au format binaire : \n",
    "import io\n",
    "# Pour la gestion des chemins de fichiers\n",
    "import os\n",
    "\n",
    "# Pour valider les résultats et les exporter en CSV\n",
    "import pandas as pd\n",
    "# Pour redimensionner les images\n",
    "from PIL import Image\n",
    "# Pour la manipulation d'arrays\n",
    "import numpy as np\n",
    "\n",
    "# Pour l'extraction de features des images\n",
    "import tensorflow as tf\n",
    "# Sur EMR AWS, avec le kernel pyspark, keras a été installé à part (lors du bootstrap)\n",
    "# \"from tensorflow.keras...\" , ne fonctionne pas; il fait faire \"from keras...\"\n",
    "from keras.applications.mobilenet_v2 import MobileNetV2, preprocess_input\n",
    "from keras.preprocessing.image import img_to_array\n",
    "from keras import Model\n",
    "\n",
    "# Création et manipulation de dataframes Spark\n",
    "from pyspark.sql.functions import col, pandas_udf, PandasUDFType, element_at, split\n",
    "\n",
    "# Pour convertion des features en vecteurs Spark\n",
    "from pyspark.ml.linalg import Vectors, VectorUDT\n",
    "from pyspark.sql.functions import udf\n",
    "# Pour réalisation de la PCA sur les features\n",
    "from pyspark.ml.feature import PCA"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **1. Définition des PATH pour charger les images et enregistrer les résultats**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "PATH = 's3://data-p8'\n",
    "PATH_Data = PATH + '/Test_med'\n",
    "PATH_Result = PATH + '/Results'\n",
    "print('PATH :        ' + PATH\\\n",
    "      + '\\nPATH_Data :   ' + PATH_Data \\\n",
    "      + '\\nPATH_Result : ' + PATH_Result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **2. Traitement des données**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **2.1 Chargement des données**\n",
    "\n",
    "Les images sont chargées au format binaire, ce qui offre plus de souplesse dans la façon de prétraiter les images.\n",
    "\n",
    "Avant de charger les images, nous spécifions que nous voulons charger uniquement les fichiers dont l'extension est **jpg**.\n",
    "\n",
    "Nous indiquons également de charger tous les objets possibles contenus dans les sous-dossiers du dossier communiqué (\"recursiveFileLookup\")."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "images = spark.read.format(\"binaryFile\") \\\n",
    "  .option(\"pathGlobFilter\", \"*.jpg\") \\\n",
    "  .option(\"recursiveFileLookup\", \"true\") \\\n",
    "  .load(PATH_Data)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Vérifions que le chargement s'est bien passé : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "images.show(5)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ajout d'une colonne contenant les **labels** de chaque image :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# split(images['path'], '/'),-2) : couper le chemin complet à chaque '/'\n",
    "# et prendre l'avant dernier élément, c'est-à-dire le nom du dossier\n",
    "# qui contient les images\n",
    "images = images.withColumn('label', element_at(split(images['path'], '/'),-2))\n",
    "print(images.printSchema())  # Affichage de la structure du dataframe\n",
    "print(images.select('path','label').show(5,False)) # True si on veut tronquer les colonnes lors de l'affichage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **2.2 Préparation du modèle**\n",
    "\n",
    "Nous allons utiliser la technique du **transfert learning** pour extraire les features des images.\n",
    "\n",
    "Nous allons utiliser le modèle **MobileNetV2** pour sa rapidité d'exécution comparée à d'autres modèles comme *VGG16* par exemple. Nous allons donc récupérer l'avant dernière couche du modèle en sortie. La dernière couche, avec sa fonction d'activation *Softmax*, est destinée à la classification, ce que nous ne souhaitons pas ici.\n",
    "\n",
    "**MobileNetV2**, lorsqu'on l'utilise en incluant toutes ses couches, attend obligatoirement des images de dimension (224,224,3). Nos images étant toutes de dimension (100,100,3), nous devrons simplement les **redimensionner** avant de les confier au modèle.\n",
    "\n",
    "<u>Dans l'odre</u> :\n",
    " 1. Nous chargeons le modèle **MobileNetV2** avec les poids **précalculés** issus d'**imagenet** et en spécifiant le format de nos images en entrée\n",
    " 2. Nous créons un nouveau modèle avec :\n",
    "  - <u>en entrée</u> : l'entrée du modèle MobileNetV2\n",
    "  - <u>en sortie</u> : l'avant dernière couche du modèle MobileNetV2"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = MobileNetV2(weights='imagenet',\n",
    "                    include_top=True,\n",
    "                    input_shape=(224, 224, 3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "new_model = Model(inputs=model.input,\n",
    "                  outputs=model.layers[-2].output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Affichage du résumé de notre nouveau modèle où nous constatons que <u>nous récupérons en sortie un vecteur de dimension (1, 1, 1280)</u> :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "new_model.summary()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Tous les workeurs doivent pouvoir accéder au modèle ainsi qu'à ses poids. <br />\n",
    "Une bonne pratique consiste à charger le modèle sur le driver puis à diffuser ensuite les poids aux différents workeurs."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "brodcast_weights = sc.broadcast(new_model.get_weights())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "<u>Mettons cela sous forme de fonction</u> :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def model_fn():\n",
    "    \"\"\"\n",
    "    Returns a MobileNetV2 model with top layer removed \n",
    "    and broadcasted pretrained weights.\n",
    "    \"\"\"\n",
    "    model = MobileNetV2(weights='imagenet',\n",
    "                        include_top=True,\n",
    "                        input_shape=(224, 224, 3))\n",
    "    for layer in model.layers:\n",
    "        layer.trainable = False\n",
    "    new_model = Model(inputs=model.input,\n",
    "                  outputs=model.layers[-2].output)\n",
    "    new_model.set_weights(brodcast_weights.value)  # Diffusion des poids\n",
    "    return new_model"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **2.3 Définition du processus de chargement des images et application de leur featurisation à travers l'utilisation de pandas UDF**\n",
    "\n",
    "Ce notebook définit la logique par étapes, jusqu'à Pandas UDF.\n",
    "\n",
    "<u>L'empilement des appels est la suivante</u> :\n",
    "\n",
    "- Pandas UDF\n",
    "  - featuriser une série d'images pd.Series\n",
    "   - prétraiter une image"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess(content):\n",
    "    \"\"\"\n",
    "    Preprocesses raw image bytes for prediction.\n",
    "    \"\"\"\n",
    "    # Chargement des images au format binaire et redimensionnement\n",
    "    img = Image.open(io.BytesIO(content)).resize([224, 224])\n",
    "    arr = img_to_array(img)  # Conversion en array\n",
    "    return preprocess_input(arr)\n",
    "\n",
    "def featurize_series(model, content_series):\n",
    "    \"\"\"\n",
    "    Featurize a pd.Series of raw images using the input model.\n",
    "    :return: a pd.Series of image features\n",
    "    \"\"\"\n",
    "    input = np.stack(content_series.map(preprocess))\n",
    "    preds = model.predict(input)\n",
    "    # Pour certaines couches, les caractéristiques de sortie seront des tenseurs multidimensionnels.\n",
    "    # Nous aplatissions les tenseurs de caractéristiques en vecteurs pour faciliter le stockage dans les DataFrames Spark.\n",
    "    output = [p.flatten() for p in preds]\n",
    "    return pd.Series(output)\n",
    "\n",
    "@pandas_udf('array<float>', PandasUDFType.SCALAR_ITER)\n",
    "def featurize_udf(content_series_iter):\n",
    "    '''\n",
    "    This method is a Scalar Iterator pandas UDF wrapping our featurization function.\n",
    "    The decorator specifies that this returns a Spark DataFrame column of type ArrayType(FloatType).\n",
    "\n",
    "    :param content_series_iter: This argument is an iterator over batches of data, where each batch\n",
    "                              is a pandas Series of image data.\n",
    "    '''\n",
    "    # Avec PandasUDFType.SCALAR_ITER, nous pouvons charger le modèle une fois puis le réutiliser\n",
    "    # pour plusieurs lots de données. Cela amortit les coûts liés au chargement des modèles volumineux.\n",
    "    model = model_fn()\n",
    "    for content_series in content_series_iter:\n",
    "        yield featurize_series(model, content_series)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **2.4 Exécution des actions d'extraction de features**\n",
    "\n",
    "Les Pandas UDF, sur de grands enregistrements (par exemple, de très grandes images), peuvent rencontrer des erreurs de type Out Of Memory (OOM).\n",
    "\n",
    "Si de telles erreurs arrivent ci-dessous, il est possible de réduire la taille du lot Arrow via 'maxRecordsPerBatch'\n",
    "\n",
    "Il ne devrait pas y avoir de problème ici, je laisse donc la commande en commentaire."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# spark.conf.set(\"spark.sql.execution.arrow.maxRecordsPerBatch\", \"1024\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Il s'agit ici d'une étape de transformation dans Spark, autrement dit l'extraction réelle des features n'aura pas encore lieu, elle sera déclenchée par une action plus tard."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Choix du nombre de partitions que l'on va créer avec \"images\"\n",
    "nb_partitions = 24\n",
    "\n",
    "features_df = images.repartition(nb_partitions).select(col(\"path\"),\n",
    "                                                       col(\"label\"),\n",
    "                                                       featurize_udf(\"content\").alias(\"features\")\n",
    "                                                      )"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Affichage de la structure du dataframe *features_df* : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "features_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **2.5 Réalisation de la PCA**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Conversion de la colonne features en vecteurs, car c'est le format d'entrée requis pour faire une PCA avec Spark : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "list_to_vector_udf = udf(lambda l: Vectors.dense(l), VectorUDT())\n",
    "features_df = features_df.withColumn('features', list_to_vector_udf('features'))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On choisit le nombre de composantes à garder avec la PCA : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "n_componants = 20"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Réalisation de la PCA, ce qui va constituer une action qui va déclencher les calculs de featurisation.\n",
    "\n",
    "En effet, pour réaliser la PCA, l'ensemble des features doit être disponible pour pouvoir calculer la matrice de covariance, les vecteurs propres et les valeurs propres associées.\n",
    "\n",
    "Par conséquent, on ne peut pas effectuer une PCA au fur et à mesure que les valeurs sont créées."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "pca = PCA(k=n_componants, inputCol=\"features\", outputCol=\"pcaFeatures\")\n",
    "model = pca.fit(features_df)\n",
    "features_df = model.transform(features_df).select(\"path\", \"label\", \"features\", \"pcaFeatures\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Affichage de la structure du dataframe *features_df* après la PCA : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "features_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **3. Écriture des résultats sous forme de fichiers**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "print(f\"Les résultats iront ici :\\n{PATH_Result}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Enregistrement des données traitées au format \"**parquet**\" :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "features_df.write.mode(\"overwrite\").parquet(PATH_Result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **4. Chargement des données enregistrées, validation des résultats et export des composantes PCA dans un fichier CSV**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **4.1 Chargement des données**\n",
    "\n",
    "On charge les données qui ont été enregistrées au format parquet dans un **DataFrame Pandas** :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df = pd.read_parquet(PATH_Result, engine='pyarrow')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On ne va garder que les noms de fichiers au lieu de tout le chemin : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Séparer les éléments de 'path' et ne garder que le dernier (nom du fichier)\n",
    "df['path'] = df['path'].apply(lambda x: x.split('/')[-1])\n",
    "# Renommer la colonne 'path' en 'filename'\n",
    "df = df.rename(columns={'path': 'filename'})"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Ici les valeurs prennent la forme d'un dictionnaire, car *features* avait été converti en vecteurs spark"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df['features'][0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous souhaitons seulement avoir les valeurs : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df['features'] = df['features'].apply(lambda x: x['values'] if x is not None else None)\n",
    "df['pcaFeatures'] = df['pcaFeatures'].apply(lambda x: x['values'] if x is not None else None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **4.2 Validation des résultats**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On valide que la dimension du vecteur de caractéristiques ('features') des images est bien de dimension 1280, c'est-à-dire la dimension de l'avant dernière couche du modèle *MobileNetV2* :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.loc[0,'features'].shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "On valide dimension du vecteur de caractéristiques après PCA ('pcaFeatures'), qui doit correspondre au nombre choisi de composantes lors de la PCA : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.loc[0,'pcaFeatures'].shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Contenu de notre dataframe Pandas : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **4.3 Export des composantes PCA dans un ficheir CSV**\n",
    "\n",
    "Enregistrement des pcaFeatures (associées aux noms de fichier et labels) au format CSV : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# np.printoptions pour éviter l'insertion de \"\\n\" dans 'pcaFeatures' dans notre fichier csv\n",
    "with np.printoptions(linewidth=10000):\n",
    "    df[['filename', 'label', 'pcaFeatures']].to_csv(PATH_Result+'/'+'pcaFeatures.csv', index=False, sep='\\t')"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.10.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 2
}