01 - Traitements (local).ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# **Réalisez un traitement dans un environnement Big Data sur le Cloud**\n",
    "\n",
    "Ce projet est basé sur le dataset [Fruits-360](https://www.kaggle.com/datasets/moltean/fruits), qui contient 90k photos détourées de fruits et légumes.\n",
    "\n",
    "Dans ce projet, nous allons procéder à deux traitements : \n",
    "- Extraction de features d'images à l'aide d'un modèle préentrainé (transfer learning).\n",
    "- Réduction de la dimensionnalité des features extraites à l'aide d'une PCA et export des résultats dans un fichier CSV.\n",
    "\n",
    "Ces traitements doivent être réalisés à l'aide de Spark, le but étant que ces derniers puissent facilement être mis à l'échelle.\n",
    "\n",
    "Ce premier notebook consiste à tester notre script PySpark en local sur un nombre réduit d'images. Un deuxième notebook sera ensuite créé pour une exécution sur le cloud."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **Import des librairies**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Mesure des durées d'exécution : \n",
    "import time\n",
    "# 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",
    "# Désactivation des messages de debugging de tensorflow.\n",
    "# Doit être exécuté avant les imports de tensorflow.\n",
    "os.environ['TF_CPP_MIN_LOG_LEVEL'] = '3'\n",
    "\n",
    "# Pour l'extraction de features des images\n",
    "import tensorflow as tf\n",
    "from tensorflow.keras.applications.mobilenet_v2 import MobileNetV2, preprocess_input\n",
    "from tensorflow.keras.preprocessing.image import img_to_array\n",
    "from tensorflow.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",
    "from pyspark.sql import SparkSession\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": [
    "### **Mesure de la durée d'exécution du notebook**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {},
   "outputs": [],
   "source": [
    "start_time = time.perf_counter()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **1. Définition des PATH pour charger les images et enregistrer les résultats**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "PATH :        /mnt/c/Users/Data Science/Desktop/OpenClassrooms/Projets/08 - Réalisez un traitement dans un environnement Big Data sur le Cloud/Projet 08 - Fichiers\n",
      "PATH_Data :   /mnt/c/Users/Data Science/Desktop/OpenClassrooms/Projets/08 - Réalisez un traitement dans un environnement Big Data sur le Cloud/Projet 08 - Fichiers/data/Test1\n",
      "PATH_Result : /mnt/c/Users/Data Science/Desktop/OpenClassrooms/Projets/08 - Réalisez un traitement dans un environnement Big Data sur le Cloud/Projet 08 - Fichiers/data/Results\n"
     ]
    }
   ],
   "source": [
    "PATH = os.getcwd()  # Dossier courant\n",
    "PATH_Data = PATH + '/data/Test1'\n",
    "PATH_Result = PATH + '/data/Results'\n",
    "print('PATH :        ' + PATH\\\n",
    "      + '\\nPATH_Data :   ' + PATH_Data \\\n",
    "      + '\\nPATH_Result : ' + PATH_Result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **2. Création de la SparkSession**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "your 131072x1 screen size is bogus. expect trouble\n",
      "Setting default log level to \"WARN\".\n",
      "To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).\n"
     ]
    }
   ],
   "source": [
    "spark = (SparkSession\n",
    "             .builder\n",
    "             .appName('P8')  # Nom de l'application\n",
    "             .master('local')  # Pour une exécution en local, local[x] si on veut utiliser x cœurs (sinon tous les dispos sont utilisés)\n",
    "             .config(\"spark.sql.parquet.writeLegacyFormat\", 'true')  # Pour pouvoir utiliser le format parquet lors de l'enregistrement des résultats\n",
    "             .getOrCreate()  # obtenir une session spark existante ou si aucune n'existe, en créer une nouvelle\n",
    ")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous créons également la variable \"**sc**\" qui est un **SparkContext** issue de la variable **spark** :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = spark.sparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Affichage des informations de Spark en cours d'execution :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "\n",
       "            <div>\n",
       "                <p><b>SparkSession - in-memory</b></p>\n",
       "                \n",
       "        <div>\n",
       "            <p><b>SparkContext</b></p>\n",
       "\n",
       "            <p><a href=\"http://172.31.188.70:4040\">Spark UI</a></p>\n",
       "\n",
       "            <dl>\n",
       "              <dt>Version</dt>\n",
       "                <dd><code>v3.5.0</code></dd>\n",
       "              <dt>Master</dt>\n",
       "                <dd><code>local</code></dd>\n",
       "              <dt>AppName</dt>\n",
       "                <dd><code>P8</code></dd>\n",
       "            </dl>\n",
       "        </div>\n",
       "        \n",
       "            </div>\n",
       "        "
      ],
      "text/plain": [
       "<pyspark.sql.session.SparkSession at 0x7f38e40f2d40>"
      ]
     },
     "execution_count": 6,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "spark"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **3. Traitement des données**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **3.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": 7,
   "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": [
    "Ajout d'une colonne contenant les **labels** de chaque image :"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- path: string (nullable = true)\n",
      " |-- modificationTime: timestamp (nullable = true)\n",
      " |-- length: long (nullable = true)\n",
      " |-- content: binary (nullable = true)\n",
      " |-- label: string (nullable = true)\n",
      "\n",
      "None\n",
      "+--------------------+------------------+\n",
      "|                path|             label|\n",
      "+--------------------+------------------+\n",
      "|file:/mnt/c/Users...|Apple Crimson Snow|\n",
      "|file:/mnt/c/Users...|Apple Crimson Snow|\n",
      "|file:/mnt/c/Users...|Apple Crimson Snow|\n",
      "|file:/mnt/c/Users...|Apple Crimson Snow|\n",
      "|file:/mnt/c/Users...|Apple Crimson Snow|\n",
      "+--------------------+------------------+\n",
      "only showing top 5 rows\n",
      "\n",
      "None\n"
     ]
    }
   ],
   "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,True)) # False si on ne veut pas tronquer les colonnes lors de l'affichage"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **3.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](https://keras.io/api/applications/mobilenet/)** 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": 9,
   "metadata": {},
   "outputs": [],
   "source": [
    "model = MobileNetV2(weights='imagenet',\n",
    "                    include_top=True,\n",
    "                    input_shape=(224, 224, 3))"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "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": 11,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Model: \"model\"\n",
      "__________________________________________________________________________________________________\n",
      " Layer (type)                Output Shape                 Param #   Connected to                  \n",
      "==================================================================================================\n",
      " input_1 (InputLayer)        [(None, 224, 224, 3)]        0         []                            \n",
      "                                                                                                  \n",
      " Conv1 (Conv2D)              (None, 112, 112, 32)         864       ['input_1[0][0]']             \n",
      "                                                                                                  \n",
      " bn_Conv1 (BatchNormalizati  (None, 112, 112, 32)         128       ['Conv1[0][0]']               \n",
      " on)                                                                                              \n",
      "                                                                                                  \n",
      " Conv1_relu (ReLU)           (None, 112, 112, 32)         0         ['bn_Conv1[0][0]']            \n",
      "                                                                                                  \n",
      " expanded_conv_depthwise (D  (None, 112, 112, 32)         288       ['Conv1_relu[0][0]']          \n",
      " epthwiseConv2D)                                                                                  \n",
      "                                                                                                  \n",
      " expanded_conv_depthwise_BN  (None, 112, 112, 32)         128       ['expanded_conv_depthwise[0][0\n",
      "  (BatchNormalization)                                              ]']                           \n",
      "                                                                                                  \n",
      " expanded_conv_depthwise_re  (None, 112, 112, 32)         0         ['expanded_conv_depthwise_BN[0\n",
      " lu (ReLU)                                                          ][0]']                        \n",
      "                                                                                                  \n",
      " expanded_conv_project (Con  (None, 112, 112, 16)         512       ['expanded_conv_depthwise_relu\n",
      " v2D)                                                               [0][0]']                      \n",
      "                                                                                                  \n",
      " expanded_conv_project_BN (  (None, 112, 112, 16)         64        ['expanded_conv_project[0][0]'\n",
      " BatchNormalization)                                                ]                             \n",
      "                                                                                                  \n",
      " block_1_expand (Conv2D)     (None, 112, 112, 96)         1536      ['expanded_conv_project_BN[0][\n",
      "                                                                    0]']                          \n",
      "                                                                                                  \n",
      " block_1_expand_BN (BatchNo  (None, 112, 112, 96)         384       ['block_1_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_1_expand_relu (ReLU)  (None, 112, 112, 96)         0         ['block_1_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_1_pad (ZeroPadding2D  (None, 113, 113, 96)         0         ['block_1_expand_relu[0][0]'] \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_1_depthwise (Depthwi  (None, 56, 56, 96)           864       ['block_1_pad[0][0]']         \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_1_depthwise_BN (Batc  (None, 56, 56, 96)           384       ['block_1_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_1_depthwise_relu (Re  (None, 56, 56, 96)           0         ['block_1_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_1_project (Conv2D)    (None, 56, 56, 24)           2304      ['block_1_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_1_project_BN (BatchN  (None, 56, 56, 24)           96        ['block_1_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_2_expand (Conv2D)     (None, 56, 56, 144)          3456      ['block_1_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_2_expand_BN (BatchNo  (None, 56, 56, 144)          576       ['block_2_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_2_expand_relu (ReLU)  (None, 56, 56, 144)          0         ['block_2_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_2_depthwise (Depthwi  (None, 56, 56, 144)          1296      ['block_2_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_2_depthwise_BN (Batc  (None, 56, 56, 144)          576       ['block_2_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_2_depthwise_relu (Re  (None, 56, 56, 144)          0         ['block_2_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_2_project (Conv2D)    (None, 56, 56, 24)           3456      ['block_2_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_2_project_BN (BatchN  (None, 56, 56, 24)           96        ['block_2_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_2_add (Add)           (None, 56, 56, 24)           0         ['block_1_project_BN[0][0]',  \n",
      "                                                                     'block_2_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_3_expand (Conv2D)     (None, 56, 56, 144)          3456      ['block_2_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_3_expand_BN (BatchNo  (None, 56, 56, 144)          576       ['block_3_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_3_expand_relu (ReLU)  (None, 56, 56, 144)          0         ['block_3_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_3_pad (ZeroPadding2D  (None, 57, 57, 144)          0         ['block_3_expand_relu[0][0]'] \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_3_depthwise (Depthwi  (None, 28, 28, 144)          1296      ['block_3_pad[0][0]']         \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_3_depthwise_BN (Batc  (None, 28, 28, 144)          576       ['block_3_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_3_depthwise_relu (Re  (None, 28, 28, 144)          0         ['block_3_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_3_project (Conv2D)    (None, 28, 28, 32)           4608      ['block_3_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_3_project_BN (BatchN  (None, 28, 28, 32)           128       ['block_3_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_4_expand (Conv2D)     (None, 28, 28, 192)          6144      ['block_3_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_4_expand_BN (BatchNo  (None, 28, 28, 192)          768       ['block_4_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_4_expand_relu (ReLU)  (None, 28, 28, 192)          0         ['block_4_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_4_depthwise (Depthwi  (None, 28, 28, 192)          1728      ['block_4_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_4_depthwise_BN (Batc  (None, 28, 28, 192)          768       ['block_4_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_4_depthwise_relu (Re  (None, 28, 28, 192)          0         ['block_4_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_4_project (Conv2D)    (None, 28, 28, 32)           6144      ['block_4_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_4_project_BN (BatchN  (None, 28, 28, 32)           128       ['block_4_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_4_add (Add)           (None, 28, 28, 32)           0         ['block_3_project_BN[0][0]',  \n",
      "                                                                     'block_4_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_5_expand (Conv2D)     (None, 28, 28, 192)          6144      ['block_4_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_5_expand_BN (BatchNo  (None, 28, 28, 192)          768       ['block_5_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_5_expand_relu (ReLU)  (None, 28, 28, 192)          0         ['block_5_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_5_depthwise (Depthwi  (None, 28, 28, 192)          1728      ['block_5_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_5_depthwise_BN (Batc  (None, 28, 28, 192)          768       ['block_5_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_5_depthwise_relu (Re  (None, 28, 28, 192)          0         ['block_5_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_5_project (Conv2D)    (None, 28, 28, 32)           6144      ['block_5_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_5_project_BN (BatchN  (None, 28, 28, 32)           128       ['block_5_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_5_add (Add)           (None, 28, 28, 32)           0         ['block_4_add[0][0]',         \n",
      "                                                                     'block_5_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_6_expand (Conv2D)     (None, 28, 28, 192)          6144      ['block_5_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_6_expand_BN (BatchNo  (None, 28, 28, 192)          768       ['block_6_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_6_expand_relu (ReLU)  (None, 28, 28, 192)          0         ['block_6_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_6_pad (ZeroPadding2D  (None, 29, 29, 192)          0         ['block_6_expand_relu[0][0]'] \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_6_depthwise (Depthwi  (None, 14, 14, 192)          1728      ['block_6_pad[0][0]']         \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_6_depthwise_BN (Batc  (None, 14, 14, 192)          768       ['block_6_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_6_depthwise_relu (Re  (None, 14, 14, 192)          0         ['block_6_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_6_project (Conv2D)    (None, 14, 14, 64)           12288     ['block_6_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_6_project_BN (BatchN  (None, 14, 14, 64)           256       ['block_6_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_7_expand (Conv2D)     (None, 14, 14, 384)          24576     ['block_6_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_7_expand_BN (BatchNo  (None, 14, 14, 384)          1536      ['block_7_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_7_expand_relu (ReLU)  (None, 14, 14, 384)          0         ['block_7_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_7_depthwise (Depthwi  (None, 14, 14, 384)          3456      ['block_7_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_7_depthwise_BN (Batc  (None, 14, 14, 384)          1536      ['block_7_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_7_depthwise_relu (Re  (None, 14, 14, 384)          0         ['block_7_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_7_project (Conv2D)    (None, 14, 14, 64)           24576     ['block_7_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_7_project_BN (BatchN  (None, 14, 14, 64)           256       ['block_7_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_7_add (Add)           (None, 14, 14, 64)           0         ['block_6_project_BN[0][0]',  \n",
      "                                                                     'block_7_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_8_expand (Conv2D)     (None, 14, 14, 384)          24576     ['block_7_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_8_expand_BN (BatchNo  (None, 14, 14, 384)          1536      ['block_8_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_8_expand_relu (ReLU)  (None, 14, 14, 384)          0         ['block_8_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_8_depthwise (Depthwi  (None, 14, 14, 384)          3456      ['block_8_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_8_depthwise_BN (Batc  (None, 14, 14, 384)          1536      ['block_8_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_8_depthwise_relu (Re  (None, 14, 14, 384)          0         ['block_8_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_8_project (Conv2D)    (None, 14, 14, 64)           24576     ['block_8_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_8_project_BN (BatchN  (None, 14, 14, 64)           256       ['block_8_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_8_add (Add)           (None, 14, 14, 64)           0         ['block_7_add[0][0]',         \n",
      "                                                                     'block_8_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_9_expand (Conv2D)     (None, 14, 14, 384)          24576     ['block_8_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_9_expand_BN (BatchNo  (None, 14, 14, 384)          1536      ['block_9_expand[0][0]']      \n",
      " rmalization)                                                                                     \n",
      "                                                                                                  \n",
      " block_9_expand_relu (ReLU)  (None, 14, 14, 384)          0         ['block_9_expand_BN[0][0]']   \n",
      "                                                                                                  \n",
      " block_9_depthwise (Depthwi  (None, 14, 14, 384)          3456      ['block_9_expand_relu[0][0]'] \n",
      " seConv2D)                                                                                        \n",
      "                                                                                                  \n",
      " block_9_depthwise_BN (Batc  (None, 14, 14, 384)          1536      ['block_9_depthwise[0][0]']   \n",
      " hNormalization)                                                                                  \n",
      "                                                                                                  \n",
      " block_9_depthwise_relu (Re  (None, 14, 14, 384)          0         ['block_9_depthwise_BN[0][0]']\n",
      " LU)                                                                                              \n",
      "                                                                                                  \n",
      " block_9_project (Conv2D)    (None, 14, 14, 64)           24576     ['block_9_depthwise_relu[0][0]\n",
      "                                                                    ']                            \n",
      "                                                                                                  \n",
      " block_9_project_BN (BatchN  (None, 14, 14, 64)           256       ['block_9_project[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_9_add (Add)           (None, 14, 14, 64)           0         ['block_8_add[0][0]',         \n",
      "                                                                     'block_9_project_BN[0][0]']  \n",
      "                                                                                                  \n",
      " block_10_expand (Conv2D)    (None, 14, 14, 384)          24576     ['block_9_add[0][0]']         \n",
      "                                                                                                  \n",
      " block_10_expand_BN (BatchN  (None, 14, 14, 384)          1536      ['block_10_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_10_expand_relu (ReLU  (None, 14, 14, 384)          0         ['block_10_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_10_depthwise (Depthw  (None, 14, 14, 384)          3456      ['block_10_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_10_depthwise_BN (Bat  (None, 14, 14, 384)          1536      ['block_10_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_10_depthwise_relu (R  (None, 14, 14, 384)          0         ['block_10_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_10_project (Conv2D)   (None, 14, 14, 96)           36864     ['block_10_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_10_project_BN (Batch  (None, 14, 14, 96)           384       ['block_10_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_11_expand (Conv2D)    (None, 14, 14, 576)          55296     ['block_10_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_11_expand_BN (BatchN  (None, 14, 14, 576)          2304      ['block_11_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_11_expand_relu (ReLU  (None, 14, 14, 576)          0         ['block_11_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_11_depthwise (Depthw  (None, 14, 14, 576)          5184      ['block_11_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_11_depthwise_BN (Bat  (None, 14, 14, 576)          2304      ['block_11_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_11_depthwise_relu (R  (None, 14, 14, 576)          0         ['block_11_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_11_project (Conv2D)   (None, 14, 14, 96)           55296     ['block_11_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_11_project_BN (Batch  (None, 14, 14, 96)           384       ['block_11_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_11_add (Add)          (None, 14, 14, 96)           0         ['block_10_project_BN[0][0]', \n",
      "                                                                     'block_11_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_12_expand (Conv2D)    (None, 14, 14, 576)          55296     ['block_11_add[0][0]']        \n",
      "                                                                                                  \n",
      " block_12_expand_BN (BatchN  (None, 14, 14, 576)          2304      ['block_12_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_12_expand_relu (ReLU  (None, 14, 14, 576)          0         ['block_12_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_12_depthwise (Depthw  (None, 14, 14, 576)          5184      ['block_12_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_12_depthwise_BN (Bat  (None, 14, 14, 576)          2304      ['block_12_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_12_depthwise_relu (R  (None, 14, 14, 576)          0         ['block_12_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_12_project (Conv2D)   (None, 14, 14, 96)           55296     ['block_12_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_12_project_BN (Batch  (None, 14, 14, 96)           384       ['block_12_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_12_add (Add)          (None, 14, 14, 96)           0         ['block_11_add[0][0]',        \n",
      "                                                                     'block_12_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_13_expand (Conv2D)    (None, 14, 14, 576)          55296     ['block_12_add[0][0]']        \n",
      "                                                                                                  \n",
      " block_13_expand_BN (BatchN  (None, 14, 14, 576)          2304      ['block_13_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_13_expand_relu (ReLU  (None, 14, 14, 576)          0         ['block_13_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_13_pad (ZeroPadding2  (None, 15, 15, 576)          0         ['block_13_expand_relu[0][0]']\n",
      " D)                                                                                               \n",
      "                                                                                                  \n",
      " block_13_depthwise (Depthw  (None, 7, 7, 576)            5184      ['block_13_pad[0][0]']        \n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_13_depthwise_BN (Bat  (None, 7, 7, 576)            2304      ['block_13_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_13_depthwise_relu (R  (None, 7, 7, 576)            0         ['block_13_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_13_project (Conv2D)   (None, 7, 7, 160)            92160     ['block_13_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_13_project_BN (Batch  (None, 7, 7, 160)            640       ['block_13_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_14_expand (Conv2D)    (None, 7, 7, 960)            153600    ['block_13_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_14_expand_BN (BatchN  (None, 7, 7, 960)            3840      ['block_14_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_14_expand_relu (ReLU  (None, 7, 7, 960)            0         ['block_14_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_14_depthwise (Depthw  (None, 7, 7, 960)            8640      ['block_14_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_14_depthwise_BN (Bat  (None, 7, 7, 960)            3840      ['block_14_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_14_depthwise_relu (R  (None, 7, 7, 960)            0         ['block_14_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_14_project (Conv2D)   (None, 7, 7, 160)            153600    ['block_14_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_14_project_BN (Batch  (None, 7, 7, 160)            640       ['block_14_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_14_add (Add)          (None, 7, 7, 160)            0         ['block_13_project_BN[0][0]', \n",
      "                                                                     'block_14_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_15_expand (Conv2D)    (None, 7, 7, 960)            153600    ['block_14_add[0][0]']        \n",
      "                                                                                                  \n",
      " block_15_expand_BN (BatchN  (None, 7, 7, 960)            3840      ['block_15_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_15_expand_relu (ReLU  (None, 7, 7, 960)            0         ['block_15_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_15_depthwise (Depthw  (None, 7, 7, 960)            8640      ['block_15_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_15_depthwise_BN (Bat  (None, 7, 7, 960)            3840      ['block_15_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_15_depthwise_relu (R  (None, 7, 7, 960)            0         ['block_15_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_15_project (Conv2D)   (None, 7, 7, 160)            153600    ['block_15_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_15_project_BN (Batch  (None, 7, 7, 160)            640       ['block_15_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " block_15_add (Add)          (None, 7, 7, 160)            0         ['block_14_add[0][0]',        \n",
      "                                                                     'block_15_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " block_16_expand (Conv2D)    (None, 7, 7, 960)            153600    ['block_15_add[0][0]']        \n",
      "                                                                                                  \n",
      " block_16_expand_BN (BatchN  (None, 7, 7, 960)            3840      ['block_16_expand[0][0]']     \n",
      " ormalization)                                                                                    \n",
      "                                                                                                  \n",
      " block_16_expand_relu (ReLU  (None, 7, 7, 960)            0         ['block_16_expand_BN[0][0]']  \n",
      " )                                                                                                \n",
      "                                                                                                  \n",
      " block_16_depthwise (Depthw  (None, 7, 7, 960)            8640      ['block_16_expand_relu[0][0]']\n",
      " iseConv2D)                                                                                       \n",
      "                                                                                                  \n",
      " block_16_depthwise_BN (Bat  (None, 7, 7, 960)            3840      ['block_16_depthwise[0][0]']  \n",
      " chNormalization)                                                                                 \n",
      "                                                                                                  \n",
      " block_16_depthwise_relu (R  (None, 7, 7, 960)            0         ['block_16_depthwise_BN[0][0]'\n",
      " eLU)                                                               ]                             \n",
      "                                                                                                  \n",
      " block_16_project (Conv2D)   (None, 7, 7, 320)            307200    ['block_16_depthwise_relu[0][0\n",
      "                                                                    ]']                           \n",
      "                                                                                                  \n",
      " block_16_project_BN (Batch  (None, 7, 7, 320)            1280      ['block_16_project[0][0]']    \n",
      " Normalization)                                                                                   \n",
      "                                                                                                  \n",
      " Conv_1 (Conv2D)             (None, 7, 7, 1280)           409600    ['block_16_project_BN[0][0]'] \n",
      "                                                                                                  \n",
      " Conv_1_bn (BatchNormalizat  (None, 7, 7, 1280)           5120      ['Conv_1[0][0]']              \n",
      " ion)                                                                                             \n",
      "                                                                                                  \n",
      " out_relu (ReLU)             (None, 7, 7, 1280)           0         ['Conv_1_bn[0][0]']           \n",
      "                                                                                                  \n",
      " global_average_pooling2d (  (None, 1280)                 0         ['out_relu[0][0]']            \n",
      " GlobalAveragePooling2D)                                                                          \n",
      "                                                                                                  \n",
      "==================================================================================================\n",
      "Total params: 2257984 (8.61 MB)\n",
      "Trainable params: 2223872 (8.48 MB)\n",
      "Non-trainable params: 34112 (133.25 KB)\n",
      "__________________________________________________________________________________________________\n"
     ]
    }
   ],
   "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": 12,
   "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": 13,
   "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": [
    "### **3.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": 14,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "/home/maxime/.local/lib/python3.10/site-packages/pyspark/sql/pandas/functions.py:407: UserWarning: In Python 3.6+ and Spark 3.0+, it is preferred to specify type hints for pandas UDF instead of specifying pandas UDF type which will be deprecated in the future releases. See SPARK-28264 for more details.\n",
      "  warnings.warn(\n"
     ]
    }
   ],
   "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": [
    "### **3.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": 15,
   "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": 16,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Choix du nombre de partitions que l'on va créer avec \"images\"\n",
    "nb_partitions = 20\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": 17,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- path: string (nullable = true)\n",
      " |-- label: string (nullable = true)\n",
      " |-- features: array (nullable = true)\n",
      " |    |-- element: float (containsNull = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "features_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **3.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": 18,
   "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": 19,
   "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": 20,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "1/1 [==============================] - 2s 2s/step                   (0 + 1) / 1]\n",
      "/opt/spark/python/lib/pyspark.zip/pyspark/sql/pandas/serializers.py:224: FutureWarning: is_categorical_dtype is deprecated and will be removed in a future version. Use isinstance(dtype, CategoricalDtype) instead\n",
      "1/1 [==============================] - 1s 539ms/step                (0 + 1) / 1]\n",
      "1/1 [==============================] - 0s 482ms/step               (0 + 1) / 20]\n",
      "1/1 [==============================] - 1s 799ms/step               (1 + 1) / 20]\n",
      "1/1 [==============================] - 0s 484ms/step               (2 + 1) / 20]\n",
      "WARNING:tensorflow:5 out of the last 5 calls to <function Model.make_predict_function.<locals>.predict_function at 0x7fd9fd522710> triggered tf.function retracing. Tracing is expensive and the excessive number of tracings could be due to (1) creating @tf.function repeatedly in a loop, (2) passing tensors with different shapes, (3) passing Python objects instead of tensors. For (1), please define your @tf.function outside of the loop. For (2), @tf.function has reduce_retracing=True option that can avoid unnecessary retracing. For (3), please refer to https://www.tensorflow.org/guide/function#controlling_retracing and https://www.tensorflow.org/api_docs/python/tf/function for  more details.\n",
      "WARNING:tensorflow:6 out of the last 6 calls to <function Model.make_predict_function.<locals>.predict_function at 0x7fd9149927a0> triggered tf.function retracing. Tracing is expensive and the excessive number of tracings could be due to (1) creating @tf.function repeatedly in a loop, (2) passing tensors with different shapes, (3) passing Python objects instead of tensors. For (1), please define your @tf.function outside of the loop. For (2), @tf.function has reduce_retracing=True option that can avoid unnecessary retracing. For (3), please refer to https://www.tensorflow.org/guide/function#controlling_retracing and https://www.tensorflow.org/api_docs/python/tf/function for  more details.\n",
      "1/1 [==============================] - 1s 573ms/step\n",
      "1/1 [==============================] - 0s 474ms/step               (4 + 1) / 20]\n",
      "1/1 [==============================] - 1s 528ms/step               (5 + 1) / 20]\n",
      "1/1 [==============================] - 0s 491ms/step               (6 + 1) / 20]\n",
      "1/1 [==============================] - 0s 453ms/step               (7 + 1) / 20]\n",
      "1/1 [==============================] - 0s 473ms/step               (8 + 1) / 20]\n",
      "1/1 [==============================] - 1s 656ms/step               (9 + 1) / 20]\n",
      "1/1 [==============================] - 0s 461ms/step              (10 + 1) / 20]\n",
      "1/1 [==============================] - 1s 606ms/step              (11 + 1) / 20]\n",
      "1/1 [==============================] - 1s 542ms/step              (12 + 1) / 20]\n",
      "1/1 [==============================] - 1s 645ms/step              (13 + 1) / 20]\n",
      "1/1 [==============================] - 0s 467ms/step              (14 + 1) / 20]\n",
      "1/1 [==============================] - 0s 465ms/step              (15 + 1) / 20]\n",
      "1/1 [==============================] - 0s 468ms/step==>           (16 + 1) / 20]\n",
      "1/1 [==============================] - 0s 474ms/step=====>        (17 + 1) / 20]\n",
      "1/1 [==============================] - 0s 474ms/step========>     (18 + 1) / 20]\n",
      "1/1 [==============================] - 0s 478ms/step===========>  (19 + 1) / 20]\n",
      "1/1 [==============================] - 0s 471ms/step                (0 + 1) / 1]\n",
      "1/1 [==============================] - 0s 474ms/step               (0 + 1) / 20]\n",
      "1/1 [==============================] - 0s 471ms/step               (1 + 1) / 20]\n",
      "1/1 [==============================] - 0s 469ms/step               (2 + 1) / 20]\n",
      "1/1 [==============================] - 0s 469ms/step               (3 + 1) / 20]\n",
      "1/1 [==============================] - 0s 474ms/step               (4 + 1) / 20]\n",
      "1/1 [==============================] - 0s 474ms/step               (5 + 1) / 20]\n",
      "1/1 [==============================] - 0s 476ms/step               (6 + 1) / 20]\n",
      "1/1 [==============================] - 0s 476ms/step               (7 + 1) / 20]\n",
      "1/1 [==============================] - 0s 468ms/step               (8 + 1) / 20]\n",
      "1/1 [==============================] - 0s 479ms/step               (9 + 1) / 20]\n",
      "1/1 [==============================] - 0s 490ms/step              (10 + 1) / 20]\n",
      "1/1 [==============================] - 0s 473ms/step              (11 + 1) / 20]\n",
      "1/1 [==============================] - 0s 480ms/step              (12 + 1) / 20]\n",
      "1/1 [==============================] - 0s 473ms/step              (13 + 1) / 20]\n",
      "1/1 [==============================] - 1s 506ms/step              (14 + 1) / 20]\n",
      "1/1 [==============================] - 0s 481ms/step>             (15 + 1) / 20]\n",
      "1/1 [==============================] - 0s 479ms/step==>           (16 + 1) / 20]\n",
      "1/1 [==============================] - 1s 506ms/step=====>        (17 + 1) / 20]\n",
      "1/1 [==============================] - 0s 473ms/step========>     (18 + 1) / 20]\n",
      "1/1 [==============================] - 0s 477ms/step===========>  (19 + 1) / 20]\n",
      "                                                                                \r"
     ]
    }
   ],
   "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": [
    "A-t-on choisi un nombre adéquat de composantes pour notre PCA ?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Ensemble, ces 20 composantes captent 91.78 % de la variance.\n"
     ]
    }
   ],
   "source": [
    "print(f\"Ensemble, ces {n_componants} composantes captent {sum(model.explainedVariance)*100:.2f} % de la variance.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Affichage de la structure du dataframe *features_df* après la PCA : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "root\n",
      " |-- path: string (nullable = true)\n",
      " |-- label: string (nullable = true)\n",
      " |-- features: vector (nullable = true)\n",
      " |-- pcaFeatures: vector (nullable = true)\n",
      "\n"
     ]
    }
   ],
   "source": [
    "features_df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **4. Écriture des résultats sous forme de fichiers**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 23,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Les résultats iront ici :\n",
      "/mnt/c/Users/Data Science/Desktop/OpenClassrooms/Projets/08 - Réalisez un traitement dans un environnement Big Data sur le Cloud/Projet 08 - Fichiers/data/Results\n"
     ]
    }
   ],
   "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": 24,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "1/1 [==============================] - 0s 487ms/step               (0 + 1) / 20]\n",
      "1/1 [==============================] - 0s 489ms/step               (1 + 1) / 20]\n",
      "1/1 [==============================] - 0s 482ms/step               (2 + 1) / 20]\n",
      "1/1 [==============================] - 0s 492ms/step               (3 + 1) / 20]\n",
      "1/1 [==============================] - 0s 476ms/step               (4 + 1) / 20]\n",
      "1/1 [==============================] - 0s 476ms/step               (5 + 1) / 20]\n",
      "1/1 [==============================] - 0s 488ms/step               (6 + 1) / 20]\n",
      "1/1 [==============================] - 0s 484ms/step               (7 + 1) / 20]\n",
      "1/1 [==============================] - 0s 478ms/step               (8 + 1) / 20]\n",
      "1/1 [==============================] - 0s 495ms/step               (9 + 1) / 20]\n",
      "1/1 [==============================] - 0s 479ms/step              (10 + 1) / 20]\n",
      "1/1 [==============================] - 0s 480ms/step              (11 + 1) / 20]\n",
      "1/1 [==============================] - 0s 477ms/step              (12 + 1) / 20]\n",
      "1/1 [==============================] - 0s 482ms/step              (13 + 1) / 20]\n",
      "1/1 [==============================] - 0s 483ms/step              (14 + 1) / 20]\n",
      "1/1 [==============================] - 0s 478ms/step>             (15 + 1) / 20]\n",
      "1/1 [==============================] - 0s 483ms/step==>           (16 + 1) / 20]\n",
      "1/1 [==============================] - 0s 492ms/step=====>        (17 + 1) / 20]\n",
      "1/1 [==============================] - 1s 501ms/step========>     (18 + 1) / 20]\n",
      "1/1 [==============================] - 1s 501ms/step===========>  (19 + 1) / 20]\n",
      "                                                                                \r"
     ]
    }
   ],
   "source": [
    "features_df.write.mode(\"overwrite\").parquet(PATH_Result)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **5. Chargement des données enregistrées, validation des résultats et export des composantes PCA dans un fichier CSV**"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **5.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": 25,
   "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": 26,
   "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": 27,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "{'type': 1,\n",
       " 'size': None,\n",
       " 'indices': None,\n",
       " 'values': array([0.        , 0.        , 0.01413861, ..., 0.        , 0.00701549,\n",
       "        0.        ])}"
      ]
     },
     "execution_count": 27,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df['features'][0]"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous souhaitons seulement avoir les valeurs : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 28,
   "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": [
    "### **5.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": 29,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(1280,)"
      ]
     },
     "execution_count": 29,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "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": 30,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "(20,)"
      ]
     },
     "execution_count": 30,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.loc[0,'pcaFeatures'].shape"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Contenu de notre dataframe Pandas : "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 31,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/html": [
       "<div>\n",
       "<style scoped>\n",
       "    .dataframe tbody tr th:only-of-type {\n",
       "        vertical-align: middle;\n",
       "    }\n",
       "\n",
       "    .dataframe tbody tr th {\n",
       "        vertical-align: top;\n",
       "    }\n",
       "\n",
       "    .dataframe thead th {\n",
       "        text-align: right;\n",
       "    }\n",
       "</style>\n",
       "<table border=\"1\" class=\"dataframe\">\n",
       "  <thead>\n",
       "    <tr style=\"text-align: right;\">\n",
       "      <th></th>\n",
       "      <th>filename</th>\n",
       "      <th>label</th>\n",
       "      <th>features</th>\n",
       "      <th>pcaFeatures</th>\n",
       "    </tr>\n",
       "  </thead>\n",
       "  <tbody>\n",
       "    <tr>\n",
       "      <th>0</th>\n",
       "      <td>r_22_100.jpg</td>\n",
       "      <td>Apple Crimson Snow</td>\n",
       "      <td>[0.0, 0.0, 0.014138607308268547, 0.0, 0.0, 0.0...</td>\n",
       "      <td>[-1.0369816809897343, -4.820961584087551, 0.52...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>1</th>\n",
       "      <td>r_41_100.jpg</td>\n",
       "      <td>Apple Crimson Snow</td>\n",
       "      <td>[0.0, 0.0, 0.0, 0.0, 0.0, 1.2477071285247803, ...</td>\n",
       "      <td>[3.746110197763563, -4.938794297068702, -1.596...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>2</th>\n",
       "      <td>r_35_100.jpg</td>\n",
       "      <td>Apple Braeburn</td>\n",
       "      <td>[0.6380921602249146, 0.035582467913627625, 0.0...</td>\n",
       "      <td>[-12.982453623476486, -5.1146858555668215, 1.6...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>3</th>\n",
       "      <td>r_42_100.jpg</td>\n",
       "      <td>Apple Braeburn</td>\n",
       "      <td>[0.8061074018478394, 0.07867594808340073, 0.0,...</td>\n",
       "      <td>[-14.861696595808477, -6.190383556400285, 3.36...</td>\n",
       "    </tr>\n",
       "    <tr>\n",
       "      <th>4</th>\n",
       "      <td>r_113_100.jpg</td>\n",
       "      <td>Apple Crimson Snow</td>\n",
       "      <td>[0.015278063714504242, 0.07300541549921036, 0....</td>\n",
       "      <td>[-11.106028177881674, -1.1916776058719598, -0....</td>\n",
       "    </tr>\n",
       "  </tbody>\n",
       "</table>\n",
       "</div>"
      ],
      "text/plain": [
       "        filename               label  \\\n",
       "0   r_22_100.jpg  Apple Crimson Snow   \n",
       "1   r_41_100.jpg  Apple Crimson Snow   \n",
       "2   r_35_100.jpg      Apple Braeburn   \n",
       "3   r_42_100.jpg      Apple Braeburn   \n",
       "4  r_113_100.jpg  Apple Crimson Snow   \n",
       "\n",
       "                                            features  \\\n",
       "0  [0.0, 0.0, 0.014138607308268547, 0.0, 0.0, 0.0...   \n",
       "1  [0.0, 0.0, 0.0, 0.0, 0.0, 1.2477071285247803, ...   \n",
       "2  [0.6380921602249146, 0.035582467913627625, 0.0...   \n",
       "3  [0.8061074018478394, 0.07867594808340073, 0.0,...   \n",
       "4  [0.015278063714504242, 0.07300541549921036, 0....   \n",
       "\n",
       "                                         pcaFeatures  \n",
       "0  [-1.0369816809897343, -4.820961584087551, 0.52...  \n",
       "1  [3.746110197763563, -4.938794297068702, -1.596...  \n",
       "2  [-12.982453623476486, -5.1146858555668215, 1.6...  \n",
       "3  [-14.861696595808477, -6.190383556400285, 3.36...  \n",
       "4  [-11.106028177881674, -1.1916776058719598, -0....  "
      ]
     },
     "execution_count": 31,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "df.head()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **5.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": 32,
   "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')"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## **6. Conclusion**\n",
    "\n",
    "Le script fonctionne en local sans difficulté. Nous allons donc pouvoir créer notre notebook qui sera exécuté sur le cloud en traitant un plus grand nombre d'images."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### **Mesure de la durée d'exécution du notebook**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 33,
   "metadata": {},
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "Durée d'exécution du notebook (hors imports) : 03:08\n"
     ]
    }
   ],
   "source": [
    "end_time = time.perf_counter()\n",
    "elapsed_time = end_time - start_time\n",
    "\n",
    "# Conversion en minutes et secondes\n",
    "minutes = int(elapsed_time // 60)\n",
    "seconds = int(elapsed_time % 60)\n",
    "\n",
    "print(f\"Durée d'exécution du notebook (hors imports) : {minutes:02}:{seconds:02}\")"
   ]
  }
 ],
 "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
}