BDF_code_final.ipynb

{
  "cells": [
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "os.environ[\"SPARK_VERSION\"] = \"spark-3.5.0\"\n",
        "!apt-get update\n",
        "!apt-get install openjdk-8-jdk-headless -qq > /dev/null\n",
        "!wget  http://apache.osuosl.org/spark/$SPARK_VERSION/$SPARK_VERSION-bin-hadoop3.tgz\n",
        "!tar xf $SPARK_VERSION-bin-hadoop3.tgz\n",
        "!echo $SPARK_VERSION-bin-hadoop3.tgz\n",
        "!rm $SPARK_VERSION-bin-hadoop3.tgz\n",
        "!pip install -q findspark"
      ],
      "metadata": {
        "id": "MF9kCZa03PvS"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "j4DhmfnW3MOW"
      },
      "outputs": [],
      "source": [
        "import os\n",
        "os.environ[\"JAVA_HOME\"] = \"/usr/lib/jvm/java-8-openjdk-amd64\"\n",
        "os.environ[\"SPARK_HOME\"] = \"/content/spark/\"\n",
        "os.environ[\"DRIVE_DATA\"]= \"/content/gdrive/Shareddrives/BDF/Graded Exercise/clusterdata-2011-2\"\n",
        "DRIVE_DATA = os.environ[\"DRIVE_DATA\"]\n",
        "\n",
        "!rm /content/spark\n",
        "!ln -s /content/$SPARK_VERSION-bin-hadoop3 /content/spark\n",
        "!export PATH=$PATH:$SPARK_HOME/bin:$SPARK_HOME/sbin\n",
        "!echo $SPARK_HOME\n",
        "!env |grep  \"DRIVE_DATA\""
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "# Mount Google Drive\n",
        "from google.colab import drive\n",
        "drive.mount('/content/gdrive')"
      ],
      "metadata": {
        "id": "BlfPxxwi3UHu"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5fVeC8Qt3MOZ",
        "outputId": "e6db3cf5-349d-4a71-ff28-9a69455c2135"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "MD5SUM\n",
            "README\n",
            "SHA1SUM\n",
            "SHA256SUM\n",
            "dominant_tasks\n",
            "job_events\n",
            "machine_attributes\n",
            "machine_events\n",
            "schema.csv\n",
            "schema.xlsx\n",
            "task_constraints\n",
            "task_events\n",
            "task_usage\n"
          ]
        }
      ],
      "source": [
        "!ls \"$DRIVE_DATA\""
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "## Imports\n"
      ],
      "metadata": {
        "id": "bFIWIaKzS8kR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "from pyspark.sql.functions import desc, avg, isnull, max\n",
        "from tqdm import tqdm\n",
        "from pyspark.sql.types import StructType, StructField, StringType, FloatType, IntegerType, BooleanType\n",
        "from pyspark.sql import SparkSession\n",
        "from pyspark.sql.functions import corr, isnull\n",
        "import matplotlib.pyplot as plt\n",
        "from pyspark.sql.functions import col, expr"
      ],
      "metadata": {
        "id": "oWLMugQ9S7xR"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "dQrWOakv3MOa"
      },
      "source": [
        "## Final working code:"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### Initialisation de la session spark"
      ],
      "metadata": {
        "id": "NMVcvdP-TXld"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Initialisation de la session Spark\n",
        "spark = SparkSession.builder \\\n",
        "    .appName(\"Top CPU Dominant Tasks\") \\\n",
        "    .master(\"local[*]\") \\\n",
        "    .getOrCreate()"
      ],
      "metadata": {
        "id": "YeTKvHdqTWSe"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "YR-6rGvf3MOc"
      },
      "source": [
        "### jobs dominants en cpu"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "suQlV8eZ3MOd"
      },
      "outputs": [],
      "source": [
        "# Chemin vers les fichiers de données\n",
        "data_path = os.path.join(os.environ[\"DRIVE_DATA\"], 'task_usage')\n",
        "\n",
        "# Générer la liste des noms de fichiers à traiter\n",
        "file_list = [f\"part-{str(i).zfill(5)}-of-00500.csv.gz\" for i in range(500)]\n",
        "file_paths = [os.path.join(data_path, filename) for filename in file_list]\n",
        "\n",
        "# Définition du schéma basé sur l'hypothétique structure des fichiers task_usage\n",
        "schema = StructType([\n",
        "    StructField(\"start time\", IntegerType(), True),\n",
        "    StructField(\"end time\", IntegerType(), True),\n",
        "    StructField(\"job ID\", StringType(), True),  # Assurez-vous que le type correspond à vos données\n",
        "    StructField(\"task index\", IntegerType(), True),\n",
        "    StructField(\"machine ID\", StringType(), True),\n",
        "    StructField(\"CPU rate\", FloatType(), True),\n",
        "    StructField(\"canonical memory usage\", FloatType(), True),\n",
        "    StructField(\"assigned memory usage\", FloatType(), True),\n",
        "    StructField(\"unmapped page cache\", FloatType(), True),\n",
        "    StructField(\"total page cache\", FloatType(), True),\n",
        "    StructField(\"maximum memory usage\", FloatType(), True),\n",
        "    StructField(\"disk I/O time\", FloatType(), True),\n",
        "    StructField(\"local disk space usage\", FloatType(), True),\n",
        "    StructField(\"maximum CPU rate\", FloatType(), True),\n",
        "    StructField(\"maximum disk IO time\", FloatType(), True),\n",
        "    StructField(\"cycles per instruction\", FloatType(), True),\n",
        "    StructField(\"memory accesses per instruction\", FloatType(), True),\n",
        "    StructField(\"sample portion\", FloatType(), True),\n",
        "    StructField(\"aggregation type\", IntegerType(), True),\n",
        "    StructField(\"sampled CPU usage\", FloatType(), True)\n",
        "])\n",
        "\n",
        "# Initialisation d'un DataFrame vide pour accumuler les résultats\n",
        "accumulated_df = None\n",
        "\n",
        "# Boucle sur chaque fichier avec tqdm pour la barre de progression\n",
        "for file in tqdm(file_paths, desc=\"Processing files\"):\n",
        "    # Vérifie si le fichier existe pour éviter les erreurs\n",
        "    if os.path.exists(file):\n",
        "        # Lecture du fichier avec le schéma défini\n",
        "        df = spark.read.csv(file, schema=schema, header=False)  # Assurez-vous que 'schema' est défini\n",
        "\n",
        "        # Filtrer les tâches où l'ID de job est NULL\n",
        "        df_filtered = df.filter(~isnull(\"job ID\"))\n",
        "\n",
        "        # Calcul des tâches dominantes en CPU pour le fichier actuel\n",
        "        top_cpu_tasks_df = df_filtered.groupBy(\"job ID\", \"task index\") \\\n",
        "                                      .agg(avg(\"CPU rate\").alias(\"average_cpu_rate\")) \\\n",
        "                                      .orderBy(desc(\"average_cpu_rate\")) \\\n",
        "                                      .limit(10)\n",
        "\n",
        "        # Accumulation des résultats\n",
        "        if accumulated_df is None:\n",
        "            accumulated_df = top_cpu_tasks_df\n",
        "        else:\n",
        "            accumulated_df = accumulated_df.union(top_cpu_tasks_df)\n",
        "\n",
        "# Après avoir accumulé les données, trier pour trouver les tâches globalement dominantes\n",
        "final_df = accumulated_df.orderBy(desc(\"average_cpu_rate\")).limit(10)\n",
        "\n",
        "# Afficher les tâches les plus dominantes en CPU à travers tous les fichiers\n",
        "# spark.stop()  # Stop the SparkContext\n",
        "# final_df_1.show()\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "C8WmIWHQ3MOe"
      },
      "outputs": [],
      "source": [
        "final_df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "6slBm1MD3MOg"
      },
      "outputs": [],
      "source": [
        "# save df_combined to a csv file\n",
        "final_df.coalesce(1).write.csv('top_cpu_tasks', header=True)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "ELbBIVpe3MOg"
      },
      "source": [
        "### jobs dominants en ram"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "XJL5Stw83MOh",
        "outputId": "2bc7171f-1554-46c5-e482-781597a3338f"
      },
      "outputs": [
        {
          "name": "stderr",
          "output_type": "stream",
          "text": [
            "Processing files: 100%|██████████| 100/100 [00:04<00:00, 24.43it/s]\n"
          ]
        }
      ],
      "source": [
        "# Définir le chemin d'accès aux fichiers et la liste des fichiers\n",
        "data_path = os.path.join(os.environ[\"DRIVE_DATA\"], 'task_usage')\n",
        "file_list = [f\"part-{str(i).zfill(5)}-of-00500.csv.gz\" for i in range(500)]\n",
        "file_paths = [os.path.join(data_path, filename) for filename in file_list]\n",
        "\n",
        "# Définir le schéma basé sur votre schéma de données\n",
        "schema = StructType([\n",
        "    StructField(\"job ID\", StringType(), True),\n",
        "    StructField(\"task index\", StringType(), True),\n",
        "    StructField(\"canonical memory usage\", FloatType(), True),\n",
        "    StructField(\"assigned memory usage\", FloatType(), True),\n",
        "    StructField(\"total page cache\", FloatType(), True),\n",
        "    StructField(\"maximum memory usage\", FloatType(), True)\n",
        "])\n",
        "\n",
        "accumulated_df = None\n",
        "\n",
        "# Boucle sur chaque fichier avec tqdm\n",
        "for file_path in tqdm(file_paths, desc=\"Processing files\"):\n",
        "    if os.path.exists(file_path):\n",
        "        # Lire le fichier\n",
        "        df = spark.read.csv(file_path, schema=schema, header=False)\n",
        "\n",
        "        # Filtrer les lignes où l'ID de job est NULL\n",
        "        df_filtered = df.filter(~isnull(\"job ID\"))\n",
        "\n",
        "        # Calculer les statistiques d'utilisation de la mémoire\n",
        "        memory_usage_stats = df_filtered.groupBy(\"job ID\", \"task index\") \\\n",
        "            .agg(\n",
        "                avg(\"canonical memory usage\").alias(\"average canonical memory usage\"),\n",
        "                max(\"canonical memory usage\").alias(\"maximum canonical memory usage\"),\n",
        "                avg(\"assigned memory usage\").alias(\"average assigned memory usage\"),\n",
        "                max(\"assigned memory usage\").alias(\"maximum assigned memory usage\"),\n",
        "                avg(\"total page cache\").alias(\"average total page cache\"),\n",
        "                max(\"total page cache\").alias(\"maximum total page cache\"),\n",
        "                avg(\"maximum memory usage\").alias(\"average maximum memory usage\"),\n",
        "                max(\"maximum memory usage\").alias(\"max maximum memory usage\")\n",
        "            ).orderBy(desc(\"max maximum memory usage\")).limit(10)\n",
        "\n",
        "        # Accumuler les résultats\n",
        "        if accumulated_df is None:\n",
        "            accumulated_df = memory_usage_stats\n",
        "        else:\n",
        "            accumulated_df = accumulated_df.union(memory_usage_stats)\n",
        "\n",
        "# Trier pour obtenir les 10 tâches globalement les plus gourmandes en mémoire\n",
        "final_dominant_memory_tasks = accumulated_df.orderBy(desc(\"max maximum memory usage\")).limit(10)\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "5L56PC9j3MOh",
        "outputId": "87dc5ca1-5e18-4634-9d2a-00c9e24e30b8"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+------------+------------+------------------------------+------------------------------+-----------------------------+-----------------------------+------------------------+------------------------+----------------------------+------------------------+\n",
            "|      job ID|  task index|average canonical memory usage|maximum canonical memory usage|average assigned memory usage|maximum assigned memory usage|average total page cache|maximum total page cache|average maximum memory usage|max maximum memory usage|\n",
            "+------------+------------+------------------------------+------------------------------+-----------------------------+-----------------------------+------------------------+------------------------+----------------------------+------------------------+\n",
            "| 93810000000| 93900000000|                  5.84456448E9|                   6.2588063E9|                       1064.5|                       8008.0|       8.9123321959375E8|             4.4860575E9|           4.570299230454566|                   145.8|\n",
            "|251999000000|252000000000|            6.24572960840404E9|                    6.280905E9|            626.4343434343434|                       4731.0|    1.3197132029343433E9|              5.140209E9|          0.3321292101756874|                   64.62|\n",
            "|105388000000|105456000000|                  5.68243488E9|                   6.2614093E9|                        185.0|                        598.0|               1303327.0|               1303327.0|           4.978990290313959|                   19.88|\n",
            "|168198000000|168300000000|           6.149899410285714E9|                    6.272648E9|           1584.6666666666667|                       4642.0|    1.6942491224761906E9|             5.1187825E9|           0.926832272752931|                   18.94|\n",
            "|186485000000|186600000000|           6.180085816615385E9|                    6.275001E9|            518.7019230769231|                       9320.0|     9.836511889038461E8|             5.3397903E9|          0.1865365973886023|                   18.88|\n",
            "| 78211000000| 78291000000|           6.202035541333333E9|                   6.2570327E9|           4047.3333333333335|                      17049.0|           3.602101504E9|             3.6021015E9|          3.1410765517891073|                   18.72|\n",
            "|272603000000|272700000000|            6.22700940055814E9|                    6.283534E9|            577.0174418604652|                       4467.0|     1.505958619860465E9|             5.3514307E9|         0.10922088732090636|                    18.0|\n",
            "|246207000000|246266000000|                6.0664900608E9|                   6.2803436E9|                        543.8|                       1867.0|               1695506.0|               1695506.0|           1.773319184128195|                   17.62|\n",
            "| 15949000000| 16200000000|           5.636060702117647E9|                    6.249154E9|           2167.9411764705883|                      14157.0|     4.253119155529412E9|             4.8201313E9|          1.0196339992056465|                   17.03|\n",
            "|256404000000|256500000000|          6.1626022725079365E9|                    6.281387E9|           1039.4603174603174|                      10378.0|    1.3244167361269841E9|              4.995305E9|         0.27873055794921203|                   16.84|\n",
            "+------------+------------+------------------------------+------------------------------+-----------------------------+-----------------------------+------------------------+------------------------+----------------------------+------------------------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Affichage des résultats\n",
        "final_dominant_memory_tasks.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "GwEtZbEI3MOj"
      },
      "source": [
        "### classification des jobs dominants par classe de priorité"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "AXyR6Aqb3MOj",
        "outputId": "c91e8bfd-84eb-4c75-d803-1fea528406ce"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+----------+----------+------------------+\n",
            "|    job ID|task index|  average_cpu_rate|\n",
            "+----------+----------+------------------+\n",
            "|6210686295|       115| 9.135018315864727|\n",
            "|5115856683|        12|  4.14795548300026|\n",
            "|6000622471|        16|1.2118503998654584|\n",
            "|6210686295|        72|1.2022168225958012|\n",
            "|6000622471|         2|1.1960032834322192|\n",
            "|6210686295|       110| 1.090427543153055|\n",
            "|6000622646|        30|1.0567158031687875|\n",
            "|6000619150|         5|1.0038247148999397|\n",
            "|6000619150|        58|0.8901584565639495|\n",
            "|6000622646|        44|0.8730517601156059|\n",
            "|4811385404|        55| 2.314707846955999|\n",
            "|6295212302|        67|1.6543118150597862|\n",
            "|6295201188|        78|1.3711346668501696|\n",
            "|6184860354|       233|0.6114499997347593|\n",
            "|6238840043|       167|0.6112993460569885|\n",
            "|6184860354|       132|0.5874687507748604|\n",
            "|6184860354|       233|0.5664588248028475|\n",
            "|6184860354|        98|0.5617941131486612|\n",
            "|6184860354|       273| 0.557211763718549|\n",
            "|6184860354|       132|0.5571000017225742|\n",
            "+----------+----------+------------------+\n",
            "only showing top 20 rows\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# import cpu dominant tasks csv F:\\hary ptor\\clutser data\\top_cpu_tasks\n",
        "top_cpu_tasks_df = spark.read.csv('top_cpu_tasks.csv', header=True, inferSchema=True)\n",
        "top_cpu_tasks_df.show()"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "OEHxH-SR3MOj"
      },
      "outputs": [],
      "source": [
        "# final_df devrait contenir les colonnes \"job ID\" et \"task index\" des jobs dominants en CPU\n",
        "task_indices = top_cpu_tasks_df.select(\"job ID\", \"task index\").distinct()\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "mnX7jM0i3MOj"
      },
      "outputs": [],
      "source": [
        "# task_indices.show(10)"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "NUcMGrMw3MOj"
      },
      "outputs": [],
      "source": [
        "# Liste des fichiers task_events (simplifiée pour l'exemple)\n",
        "task_event_files = [f'part-{i:05d}-of-00500.csv.gz' for i in range(500)]\n",
        "\n",
        "task_events_schema = StructType([\n",
        "    StructField(\"timestamp\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"missing info\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"job ID\", StringType(), True),  # Assurez-vous que ce type correspond à celui utilisé dans final_df\n",
        "    StructField(\"task index\", StringType(), True),  # Assurez-vous que ce type correspond\n",
        "    StructField(\"machine ID\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"event type\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"user\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"scheduling class\", StringType(), True),  # Exemple, ajustez selon votre besoin\n",
        "    StructField(\"priority\", IntegerType(), True),  # La priorité, essentielle pour votre analyse\n",
        "    # Ajoutez d'autres champs selon le schéma de votre fichier task_events\n",
        "])\n",
        "\n",
        "# Initialiser un DataFrame pour accumuler les résultats\n",
        "accumulated_priority_df = None\n",
        "\n",
        "for file_name in task_event_files:\n",
        "    file_path = os.path.join(os.environ[\"DRIVE_DATA\"], 'task_events', file_name)\n",
        "    if os.path.exists(file_path):\n",
        "        df_temp = spark.read.csv(file_path, schema=task_events_schema, header=False)\n",
        "\n",
        "        # Filtrer df_temp pour garder seulement les lignes où \"task index\" est dans task_indices\n",
        "        # Ceci est une simplification; en réalité, vous auriez besoin d'une jointure ici\n",
        "        filtered_df = df_temp.join(task_indices, [\"job ID\", \"task index\"], \"inner\")\n",
        "\n",
        "        # Accumuler\n",
        "        if accumulated_priority_df is None:\n",
        "            accumulated_priority_df = filtered_df\n",
        "        else:\n",
        "            accumulated_priority_df = accumulated_priority_df.union(filtered_df)\n",
        "\n",
        "# À ce stade, accumulated_priority_df contient les priorités pour les indices de tâches de final_df\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "Kh6gEYMA3MOj",
        "outputId": "b3bccd45-5c56-41f9-8483-1b73173991ab"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+----------+----------+-------------+------------+----------+----------+--------------------+----------------+--------+\n",
            "|    job ID|task index|    timestamp|missing info|machine ID|event type|                user|scheduling class|priority|\n",
            "+----------+----------+-------------+------------+----------+----------+--------------------+----------------+--------+\n",
            "|6272076905|         0| 163027215040|        NULL|      NULL|         0|Gx2a4JlY7sTN3Jiqp...|               3|      10|\n",
            "|6272076905|         0|1347760079184|        NULL|4820073668|         3|Gx2a4JlY7sTN3Jiqp...|               3|      10|\n",
            "|6272076905|         0| 163059602676|        NULL|4820073668|         1|Gx2a4JlY7sTN3Jiqp...|               3|      10|\n",
            "|6272076905|         0|1347760079195|        NULL|      NULL|         0|Gx2a4JlY7sTN3Jiqp...|               3|      10|\n",
            "|6272076905|         0|1347790444482|        NULL| 257347123|         1|Gx2a4JlY7sTN3Jiqp...|               3|      10|\n",
            "|6295212302|        67|1976031581791|        NULL|1334665910|         5|tjYrEWcPX1rXEj2HG...|               2|       9|\n",
            "|6400512805|        15|1738692084650|        NULL|4802098790|         5|oPxcKd7feXmw+sZKr...|               0|       9|\n",
            "|6184860354|        93|1555812792691|        NULL|4821070690|         5|y9NKAC+Ud/LMV2fx5...|               2|       9|\n",
            "|4811385404|        55|1613481994734|        NULL|3889402292|         5|rNyxTd1B3RnDJBIof...|               3|       9|\n",
            "|6184860354|       273|1555813485370|        NULL|4820235500|         5|y9NKAC+Ud/LMV2fx5...|               2|       9|\n",
            "+----------+----------+-------------+------------+----------+----------+--------------------+----------------+--------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# accumulated_priority_df a une colonne \"priority\"\n",
        "top_priority_jobs = accumulated_priority_df.orderBy(col(\"priority\").desc()).limit(10)\n",
        "top_priority_jobs.show()\n"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "lEaNrgu73MOj"
      },
      "source": [
        "### etude de la corrélation entre la consommation de cpu et de la mémoire vive"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "CjxAe6g53MOk",
        "outputId": "d081a596-2f0d-410c-d8fe-3a6650771b64"
      },
      "outputs": [
        {
          "name": "stderr",
          "output_type": "stream",
          "text": [
            "Processing files: 100%|██████████| 500/500 [00:18<00:00, 26.62it/s]\n"
          ]
        }
      ],
      "source": [
        "# Définir le schéma basé sur votre structure de données\n",
        "schema = StructType([\n",
        "    StructField(\"start time\", StringType(), True),\n",
        "    StructField(\"end time\", StringType(), True),\n",
        "    StructField(\"job ID\", StringType(), True),\n",
        "    StructField(\"task index\", StringType(), True),\n",
        "    StructField(\"machine ID\", StringType(), True),\n",
        "    StructField(\"CPU rate\", FloatType(), True),\n",
        "    StructField(\"canonical memory usage\", FloatType(), True),\n",
        "    StructField(\"assigned memory usage\", FloatType(), True),\n",
        "    StructField(\"unmapped page cache\", FloatType(), True),\n",
        "    StructField(\"total page cache\", FloatType(), True),\n",
        "    StructField(\"maximum memory usage\", FloatType(), True),\n",
        "    StructField(\"disk I/O time\", FloatType(), True),\n",
        "    StructField(\"local disk space usage\", FloatType(), True),\n",
        "    StructField(\"maximum CPU rate\", FloatType(), True),\n",
        "    StructField(\"maximum disk IO time\", FloatType(), True),\n",
        "    StructField(\"cycles per instruction\", FloatType(), True),\n",
        "    StructField(\"memory accesses per instruction\", FloatType(), True),\n",
        "    StructField(\"sample portion\", FloatType(), True),\n",
        "    StructField(\"aggregation type\", StringType(), True),\n",
        "    StructField(\"sampled CPU usage\", FloatType(), True)\n",
        "])\n",
        "\n",
        "# Chemin d'accès aux fichiers et liste des fichiers\n",
        "data_path = os.path.join(os.environ[\"DRIVE_DATA\"], 'task_usage')\n",
        "file_list = [f\"part-{str(i).zfill(5)}-of-00500.csv.gz\" for i in range(500)]\n",
        "file_paths = [os.path.join(data_path, filename) for filename in file_list]\n",
        "\n",
        "# Accumuler les données nécessaires pour le calcul de corrélation\n",
        "accumulated_data_for_correlation = None\n",
        "\n",
        "# Boucle sur chaque fichier avec tqdm\n",
        "for file_path in tqdm(file_paths, desc=\"Processing files\"):\n",
        "    if os.path.exists(file_path):\n",
        "        # Lire le fichier\n",
        "        df = spark.read.csv(file_path, schema=schema, header=False)\n",
        "\n",
        "        # Filtrer les lignes où l'ID de job est NULL et sélectionner les colonnes nécessaires\n",
        "        filtered_df = df.filter(~isnull(\"job ID\")).select(\"CPU rate\", \"canonical memory usage\", \"assigned memory usage\")\n",
        "\n",
        "        # Accumuler les données\n",
        "        if accumulated_data_for_correlation is None:\n",
        "            accumulated_data_for_correlation = filtered_df\n",
        "        else:\n",
        "            accumulated_data_for_correlation = accumulated_data_for_correlation.union(filtered_df)\n",
        "\n",
        "# Calculer le coefficient de corrélation sur les données accumulées\n",
        "correlation_cpu_memory = accumulated_data_for_correlation.select(\n",
        "    corr(\"CPU rate\", \"canonical memory usage\").alias(\"correlation_cpu_canonical_memory\"),\n",
        "    corr(\"CPU rate\", \"assigned memory usage\").alias(\"correlation_cpu_assigned_memory\")\n",
        ")\n",
        "\n",
        "\n"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "-3HUKGdL3MOk",
        "outputId": "10ec27bc-8ec5-4adb-ccd4-707e7160c267"
      },
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "+--------------------------------+-------------------------------+\n",
            "|correlation_cpu_canonical_memory|correlation_cpu_assigned_memory|\n",
            "+--------------------------------+-------------------------------+\n",
            "|               0.352491425557883|           0.002450362193947116|\n",
            "+--------------------------------+-------------------------------+\n",
            "\n"
          ]
        }
      ],
      "source": [
        "# Afficher le coefficient de corrélation\n",
        "correlation_cpu_memory.show()"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "### analyse de la durée des jobs et des tâches"
      ],
      "metadata": {
        "id": "PYxCRKp3SZLI"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "# Définir le schéma pour la lecture des données\n",
        "schema = StructType([\n",
        "    StructField(\"start time\", StringType(), True),\n",
        "    StructField(\"end time\", StringType(), True),\n",
        "    StructField(\"job ID\", StringType(), True),\n",
        "    StructField(\"task index\", StringType(), True),\n",
        "    StructField(\"machine ID\", StringType(), True),\n",
        "    StructField(\"CPU rate\", FloatType(), True),\n",
        "    StructField(\"canonical memory usage\", FloatType(), True),\n",
        "    StructField(\"assigned memory usage\", FloatType(), True),\n",
        "    # Ajoutez les autres champs selon la structure de vos données\n",
        "])\n",
        "\n",
        "# Chemin d'accès aux fichiers\n",
        "data_path = os.path.join(os.environ[\"DRIVE_DATA\"], 'task_usage')\n",
        "file_list = [f\"part-{str(i).zfill(5)}-of-00500.csv.gz\" for i in range(4)]\n",
        "file_paths = [os.path.join(data_path, filename) for filename in file_list]\n",
        "\n",
        "# Accumuler les durées de toutes les tâches\n",
        "accumulated_durations = None\n",
        "\n",
        "for file_path in tqdm(file_paths, desc=\"Processing files\"):\n",
        "    if os.path.exists(file_path):\n",
        "        df = spark.read.csv(file_path, schema=schema, header=False)\n",
        "        # Calculer la durée et la convertir en secondes\n",
        "        durations = df.withColumn(\"duration\", (col(\"end time\").cast(\"long\") - col(\"start time\").cast(\"long\")) / 1000000)\n",
        "\n",
        "        if accumulated_durations is None:\n",
        "            accumulated_durations = durations\n",
        "        else:\n",
        "            accumulated_durations = accumulated_durations.union(durations)\n",
        "\n",
        "# Après l'accumulation, analyser la distribution de la durée des tâches\n",
        "accumulated_durations.describe(\"duration\").show()\n",
        "\n",
        "# Calcul des percentiles pour la durée\n",
        "percentiles = accumulated_durations.approxQuantile(\"duration\", [0.25, 0.5, 0.75], 0)\n",
        "\n",
        "# Calcul de l'IQR pour détecter les outliers\n",
        "IQR = percentiles[2] - percentiles[0]\n",
        "lower_bound = percentiles[0] - 1.5 * IQR\n",
        "upper_bound = percentiles[2] + 1.5 * IQR\n",
        "\n",
        "# Filtrer les tâches considérées comme outliers\n",
        "outliers = accumulated_durations.filter((col(\"duration\") < lower_bound) | (col(\"duration\") > upper_bound))\n",
        "outliers.show()\n"
      ],
      "metadata": {
        "id": "3femkxlOSY32"
      },
      "execution_count": null,
      "outputs": []
    }
  ],
  "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.11"
    },
    "colab": {
      "provenance": []
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}