fr/TP5/tp5.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# TP 5\n",
    "\n",
    "## Goals\n",
    "\n",
    "### Utilisation de pyspark\n",
    "* filter, map and reduce\n",
    "* Utilisation des fichiers CSV et JSON\n",
    "* SQL et pyspark\n",
    "* Écriture et lecture de fichiers ORC et Parquet"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercice 1 \\[★\\]\n",
    "\n",
    "Tout d'abord, nous allons installer pyspark\n",
    "\n",
    "`$ pip install pyspark`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Si l'installation a réussi, le code suivant sera exécuté avec succès."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark import SparkConf\n",
    "from pyspark.context import SparkContext"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous allons utiliser la configuration par défaut pour créer un contexte spark.\n",
    "Ce contexte spark sera utilisé tout au long de nos exemples.\n",
    "\n",
    "Tout d'abord, nous lisons un fichier CSV qui contient deux colonnes : languageLabel et year."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate(SparkConf())\n",
    "lines = sc.textFile(\"../../data/pl.csv\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous allons utiliser les méthodes map() et reduce() avec les expressions lambda.\n",
    "\n",
    "Notez que dans l'exercice suivant, nous n'utilisons pas les fonctions intégrées de Python.\n",
    "Le but de cet exemple est de compter le nombre de caractères ligne par ligne."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "line_length = lines.map(lambda line: len(line))\n",
    "total_length = line_length.reduce(lambda a, b: a + b)\n",
    "print(total_length)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dans cet exemple, nous obtenons le nombre de tokens séparés par des virgules pour chaque ligne.\n",
    "Enfin, nous calculons le nombre total de tokens."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "line_token_count = lines.map(lambda line: len(line.split(\",\")))\n",
    "total_token_count = line_token_count.reduce(lambda a, b: a + b)\n",
    "print(total_token_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous modifions légèrement le code et utilisons l'espace comme séparateur pour compter le nombre de tokens."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "line_token_count = lines.map(lambda line: len(line.split(\" \")))\n",
    "total_token_count = line_token_count.reduce(lambda a, b: a + b)\n",
    "print(total_token_count)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Jusqu'à présent, nous n'avons considéré qu'un seul fichier à la fois.\n",
    "\n",
    "Que faire si nous voulons manipuler et analyser plusieurs fichiers CSV?\n",
    "\n",
    "La méthode textFile() peut également gérer de tels cas."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate(SparkConf())\n",
    "lines = sc.textFile(\"../../data/*.csv\")\n",
    "\n",
    "line_length = lines.map(lambda line: len(line))\n",
    "total_length = line_length.reduce(lambda a, b: a + b)\n",
    "print(total_length)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Notre prochain objectif est de collecter toutes les lignes stockées dans plusieurs fichiers et d'y accéder en utilisant une seule variable.\n",
    "\n",
    "Ceci est possible en utilisant la méthode collect()."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sc = SparkContext.getOrCreate(SparkConf())\n",
    "lines = sc.textFile(\"../../data/*.csv\")\n",
    "\n",
    "all_lines = lines.collect()\n",
    "print(len(all_lines))\n",
    "for line in all_lines:\n",
    "    print(line)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comme la fonction intégrée de Python map(), \n",
    "il est également possible de passer une fonction définie par l'utilisateur comme entrée de la fonction map()\n",
    "de pyspark.\n",
    "\n",
    "Dans l'exemple suivant, nous comptons le nombre d'occurrences d'un mot particulier dans tous les fichiers CSV.\n",
    "\n",
    "Dans ce but, nous avons écrit une fonction count_Language() qui compte le nombre d'occurrences de \"Language\" dans une ligne. \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import re\n",
    "\n",
    "sc = SparkContext.getOrCreate(SparkConf())\n",
    "lines = sc.textFile(\"../../data/*.csv\")\n",
    "\n",
    "\n",
    "def count_Language(line):\n",
    "    return len(re.findall(\"Language\", line))\n",
    "\n",
    "\n",
    "line_length = lines.map(count_Language)\n",
    "total_length = line_length.reduce(lambda a, b: a + b)\n",
    "print(total_length)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Question** Téléchargez 50 pages HTML. Écrivez un programme utilisant Spark pour compter le nombre total de fois où `<div>` ou `<div/>` est présent dans tous ces fichiers téléchargés."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercice 2 \\[★★\\]\n",
    "\n",
    "Notre prochain objectif est de travailler avec des fichiers JSON et de créer des dataframes pandas.\n",
    "\n",
    "Cependant, nous aimerions utiliser le support des dataframes fourni par Spark.\n",
    "\n",
    "Nous créons d'abord une session Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql import SparkSession\n",
    "\n",
    "spark = SparkSession.builder.getOrCreate()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous lisons un fichier JSON et le convertissons en dataframe pandas.\n",
    "\n",
    "Ce dataframe pandas est ensuite converti en dataframe Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pandas import json_normalize\n",
    "import pandas as pd\n",
    "import json\n",
    "\n",
    "data = json.load(open(\"../../data/pl.json\"))\n",
    "dataframe = json_normalize(data)\n",
    "dataframe = dataframe.astype(dtype={\"languageLabel\": \"<U200\", \"year\": \"int64\"})\n",
    "\n",
    "df = spark.createDataFrame(dataframe)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous affichons maintenant le dataframe Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Il est également possible de spécifier le nombre de lignes à afficher."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.show(30)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Le code suivant affichera le schéma du dataframe."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.printSchema()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dans le code suivant, nous écrivons une fonction filtre pour obtenir les langages de programmation\n",
    "qui ont été publiés en 1952."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def pandas_filter_func(iterator):\n",
    "    for pandas_df in iterator:\n",
    "        yield pandas_df[pandas_df.year == 1952]\n",
    "\n",
    "\n",
    "df.mapInPandas(pandas_filter_func, schema=df.schema).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Dans le code suivant, nous appliquons une fonction définie par l'utilisateur sur la colonne année pour obtenir le siècle où un langage de programmation a été publié pour la première fois."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyspark.sql.functions import pandas_udf\n",
    "\n",
    "\n",
    "@pandas_udf(\"long\")\n",
    "def century(series: pd.Series) -> pd.Series:\n",
    "    return (series / 100) + 1\n",
    "\n",
    "\n",
    "df.select(century(df.year)).show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Comme pour pandas, il est possible de regrouper les données par valeurs de colonnes spécifiques.\n",
    "\n",
    "Dans l'exemple suivant, nous voulons obtenir le nombre de langages de programmation publiés pour la première fois chaque année."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.groupby(\"year\").count().show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Nous avons jusqu'à présent utilisé des fichiers CSV et JSON.\n",
    "\n",
    "Mais pour une performance optimisée, les fichiers ORC et parquet sont suggérés.\n",
    "\n",
    "Dans les exemples suivants, nous voyons comment nous pouvons écrire des fichiers ORC et parquet en utilisant un dataframe Spark."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.write.orc(\"languages.orc\")\n",
    "spark.read.orc(\"languages.orc\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.write.parquet(\"languages.parquet\")\n",
    "spark.read.parquet(\"languages.parquet\").show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.write.csv(\"languages.csv\")\n",
    "spark.read.csv(\"languages.csv\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "Avant de continuer, vérifiez votre répertoire actuel et voyez comment ces dataframes ont été écrits."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Question** Écrire une requête sur Wikidata pour télécharger les noms de tous les logiciels ainsi que leur première date de sortie. Écrivez un programme utilisant Spark pour obtenir le nombre de logiciels sortis chaque année."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Exercice 3 \\[★★\\]\n",
    "\n",
    "Il est possible de travailler en utilisant le langage SQL sur des dataframes spark.\n",
    "\n",
    "Pour cela, nous allons créer des vues temporaires et exécuter des requêtes SQL.\n",
    "\n",
    "L'exemple suivant va retourner et afficher toutes les langues."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "df.createOrReplaceTempView(\"languages\")\n",
    "spark.sql(\"SELECT * from languages\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "L'exemple suivant utilise une requête SQL pour obtenir le nombre de langages de programmation."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT count(*) from languages\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La requête suivante renvoie le nombre de langues publiées pour la première fois en 1952."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT count(*) from languages where year=1952\").show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "La requête suivante est similaire à l'exemple groupby vu précédemment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "spark.sql(\"SELECT year, count(*) from languages Group by year ORDER by year\").show(100)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "**Question** À la place des fonctions d'agrégation de Spark, utilisez SQL et Spark pour obtenir le nombre de logiciels sortis chaque année en utilisant les données téléchargées dans l'exercice précédent."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "For more examples, \n",
    "check [official document](https://spark.apache.org/docs/latest/api/python/getting_started/quickstart.html)"
   ]
  }
 ],
 "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.8.10"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}