<jupyter_start><jupyter_text>TP Spark - Utilisation des DataFrames  Présentation Le projet open data de la ville de San Francisco (The SF OpenData project) a été lancé en 2009 et contient des centaines de datasets concernant la ville et l'agglomération de San Francisco.Dans ce TP nous allons analyser les appels aux pompiers (SF Fire Department), à la recherche de réponses pour les questions suivantes :* Combien de types différents d'appel ont été enregistrés ?* Combien d'incidents de chaque type ont été recensés ?* Combien d'années sont enregistrées dans le fichier ?* Combien d'appels ont été enregistrés lors d'une semaine donnée ? * Quel quartier a généré plus d'appels en 2004 ?  Téléchargement du Dataset<jupyter_code>!mkdir dataset
!wget http://cosy.univ-reims.fr/~lsteffenel/cours/Master2/RT0902-BigData/Fire_Department_Calls_for_Service.csv -O dataset/Fire_Department_Calls_for_Service.csv<jupyter_output><empty_output><jupyter_text>Lecture des donnéesPour commencer, nous allons lire une base de données déjà sur HDFS.Cette base est en format CSV. Au lieu de parser nous mêmes le fichier, nous allons demander à Spark de faire la lecture, de parser le fichier et même de deviner le type de donnée des colonnes.Observez aussi l'option `header=True` qui indique à Spark de ne pas inclure la première ligne de la table (mais de l'utiliser pour le nom des colonnes).<jupyter_code>import pyspark
import random
from pyspark.sql import SparkSession
spark = SparkSession.builder.master("local").appName('TP Datasets Spark').getOrCreate()
spark.conf.set("spark.sql.repl.eagerEval.enabled", True)
%time fireServiceCallsDF = spark.read.csv("/home/jovyan/dataset/Fire_Department_Calls_for_Service.csv", header=True, inferSchema=True)<jupyter_output><empty_output><jupyter_text>L'opération précédente a pris du temps non à cause de la taille du fichier (presque 400MB) mais pour une raison plus bête : * Spark a dû deviner le schéma (le type de données) de chaque colonne. Pour cela, il a lu le fichier afin de vérifier si les types devinés ne sont pas en conflit (par exemple, supposer qu'une colonne est composée d'entiers mais vers la fin on retrouve des string)Une manière d'accélérer ce processus est de fournir le schéma (si on le connaît), ce qui a aussi l'avantage d'être plus précis :<jupyter_code>from pyspark.sql.types import StructType, StructField, IntegerType, StringType, BooleanType
fireSchema = StructType([StructField('CallNumber', IntegerType(), True),
                     StructField('UnitID', StringType(), True),
                     StructField('IncidentNumber', IntegerType(), True),
                     StructField('CallType', StringType(), True),                  
                     StructField('CallDate', StringType(), True),       
                     StructField('WatchDate', StringType(), True),       
                     StructField('ReceivedDtTm', StringType(), True),       
                     StructField('EntryDtTm', StringType(), True),       
                     StructField('DispatchDtTm', StringType(), True),       
                     StructField('ResponseDtTm', StringType(), True),       
                     StructField('OnSceneDtTm', StringType(), True),       
                     StructField('TransportDtTm', StringType(), True),                  
                     StructField('HospitalDtTm', StringType(), True),       
                     StructField('CallFinalDisposition', StringType(), True),       
                     StructField('AvailableDtTm', StringType(), True),       
                     StructField('Address', StringType(), True),       
                     StructField('City', StringType(), True),       
                     StructField('ZipcodeofIncident', IntegerType(), True),       
                     StructField('Battalion', StringType(), True),                 
                     StructField('StationArea', StringType(), True),       
                     StructField('Box', StringType(), True),       
                     StructField('OriginalPriority', StringType(), True),       
                     StructField('Priority', StringType(), True),       
                     StructField('FinalPriority', IntegerType(), True),       
                     StructField('ALSUnit', BooleanType(), True),       
                     StructField('CallTypeGroup', StringType(), True),
                     StructField('NumberofAlarms', IntegerType(), True),
                     StructField('UnitType', StringType(), True),
                     StructField('Unitsequenceincalldispatch', IntegerType(), True),
                     StructField('FirePreventionDistrict', StringType(), True),
                     StructField('SupervisorDistrict', StringType(), True),
                     StructField('NeighborhoodDistrict', StringType(), True),
                     StructField('Location', StringType(), True),
                     StructField('RowID', StringType(), True)])

# Observez comment ça va plus vite ! Comme il ne faut plus deviner (en regardant le fichier), Spark accepte le schéma sans perdre du temps 
# Attention : comme Python est un langage non-typé, nous n'obtenons que des DataFrame (les Datasets, plus avancés encore, sont disponibles pour Scala)

%time fireServiceCallsDF = spark.read.csv("/home/jovyan/dataset/Fire_Department_Calls_for_Service.csv", header=True, schema=fireSchema)
# Ici on a un extrait des 5 premières lignes du fichier.
sample=fireServiceCallsDF.take(5)
sample
#On peut vérifier le schéma avec la méthode "printSchema()"


fireServiceCallsDF.printSchema()
# il est aussi possible d'imprimer juste le nom des colonnes

fireServiceCallsDF.columns
# combien de lignes il y a dans le DataFrame ? (ça devrait afficher 1190109 entrées)
# Ça prend beaucoup de temps car il faut quand même parcourir tout le fichier.

fireServiceCallsDF.count()<jupyter_output><empty_output><jupyter_text>Ouvrez la documentation de Spark 2.0 dans un autre onglet, afin de pouvoir consulter rapidement l'API :1. Spark 2.0 docs: http://spark.apache.org/docs/latest/api/python/index.html2. DataFrame user documentation: http://spark.apache.org/docs/latest/sql-programming-guide.html3. PySpark API 2.0 docs: http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.htmlpyspark.sql.DataFrame Les DataFrames supportent deux types d'operations : les transformations et les actions.![Transfos](http://cosy.univ-reims.fr/~lsteffenel/images/trans_and_actions.png "Transformations et actions")Les transformations comme select() ou filter() créent un nouveau DataFrame à partir d'un DataFrame existant.Les actions telles que show() ou count() effectuent une action qui résulte en un résultat retourné à l'utilisateur. Certaines actions comme save() sont utilisées pour écrire le DataFrame dans le système de stockage distribué.**Les transformations contribuent à la planification de la requête, mais rien d'est vraiment executé jusqu'à ce qu'une action soit appelée.**   Question 1 : Combien de types différents d’appel ont été enregistrés ? Listez tous les différents types.Astuce : Pour répondre à cette question, utilisez les méthodes `select()` et `distinct()`.<jupyter_code># dans cet exemple on affiche juste les 5 premiers types
fireServiceCallsDF.select('CallType').show(5)

# combient de types différents ? Affichez juste une fois chacun des types
# votre code ici<jupyter_output><empty_output><jupyter_text>Question 2 : Combien d’incidents de chaque type ont été recensés ?Astuce : Utilisez `groupBy()` et `count()`, voir même `orderBy()` pour trier l'affichage.<jupyter_code># votre code ici<jupyter_output><empty_output><jupyter_text>Question 3 : Combien d’années sont enregistrées dans le fichier ?Si vous faites attention, les colonnes relatives aux dates et horaires sont interprétées comme des string et pas comme des dates ou timestamps.<jupyter_code>`fireServiceCallsDF.printSchema()`
root
 |-- CallNumber: integer (nullable = true)
 |-- UnitID: string (nullable = true)
 |-- IncidentNumber: integer (nullable = true)
 |-- CallType: string (nullable = true)
 |-- CallDate: string (nullable = true)
 |-- WatchDate: string (nullable = true)
 |-- ReceivedDtTm: string (nullable = true)
 |-- EntryDtTm: string (nullable = true)
 |-- DispatchDtTm: string (nullable = true)
 |-- ResponseDtTm: string (nullable = true)
 |-- OnSceneDtTm: string (nullable = true)
 |-- TransportDtTm: string (nullable = true)
 |-- HospitalDtTm: string (nullable = true)
 |-- CallFinalDisposition: string (nullable = true)
 |-- AvailableDtTm: string (nullable = true)
 |-- Address: string (nullable = true)
 |-- City: string (nullable = true)
 |-- ZipcodeofIncident: integer (nullable = true)
 |-- Battalion: string (nullable = true)
 |-- StationArea: string (nullable = true)
 |-- Box: string (nullable = true)
 |-- OriginalPriority: string (nullable = true)
 |-- Priority: string (nullable = true)
 |-- FinalPriority: integer (nullable = true)
 |-- ALSUnit: boolean (nullable = true)
 |-- CallTypeGroup: string (nullable = true)
 |-- NumberofAlarms: integer (nullable = true)
 |-- UnitType: string (nullable = true)
 |-- Unitsequenceincalldispatch: integer (nullable = true)
 |-- FirePreventionDistrict: string (nullable = true)
 |-- SupervisorDistrict: string (nullable = true)
 |-- NeighborhoodDistrict: string (nullable = true)
 |-- Location: string (nullable = true)
 |-- RowID: string (nullable = true)<jupyter_output><empty_output><jupyter_text>Nous pouvons utiliser la fonction `unix_timestamp()` afin de convertir les string en timestamp:[http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.htmlpyspark.sql.functions.from_unixtime](http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.htmlpyspark.sql.functions.from_unixtime)<jupyter_code>from pyspark.sql.functions import *
from_pattern1 = 'MM/dd/yyyy'
to_pattern1 = 'yyyy-MM-dd'

from_pattern2 = 'MM/dd/yyyy hh:mm:ss aa'
to_pattern2 = 'MM/dd/yyyy hh:mm:ss aa'

# Création d'un nouveau DataFrame qui utilise les dates converties en timestamp
fireServiceCallsTsDF = fireServiceCallsDF \
  .withColumn('CallDateTS', unix_timestamp(fireServiceCallsDF['CallDate'], from_pattern1).cast("timestamp")) \
  .drop('CallDate') \
  .withColumn('WatchDateTS', unix_timestamp(fireServiceCallsDF['WatchDate'], from_pattern1).cast("timestamp")) \
  .drop('WatchDate') \
  .withColumn('ReceivedDtTmTS', unix_timestamp(fireServiceCallsDF['ReceivedDtTm'], from_pattern2).cast("timestamp")) \
  .drop('ReceivedDtTm') \
  .withColumn('EntryDtTmTS', unix_timestamp(fireServiceCallsDF['EntryDtTm'], from_pattern2).cast("timestamp")) \
  .drop('EntryDtTm') \
  .withColumn('DispatchDtTmTS', unix_timestamp(fireServiceCallsDF['DispatchDtTm'], from_pattern2).cast("timestamp")) \
  .drop('DispatchDtTm') \
  .withColumn('ResponseDtTmTS', unix_timestamp(fireServiceCallsDF['ResponseDtTm'], from_pattern2).cast("timestamp")) \
  .drop('ResponseDtTm') \
  .withColumn('OnSceneDtTmTS', unix_timestamp(fireServiceCallsDF['OnSceneDtTm'], from_pattern2).cast("timestamp")) \
  .drop('OnSceneDtTm') \
  .withColumn('TransportDtTmTS', unix_timestamp(fireServiceCallsDF['TransportDtTm'], from_pattern2).cast("timestamp")) \
  .drop('TransportDtTm') \
  .withColumn('HospitalDtTmTS', unix_timestamp(fireServiceCallsDF['HospitalDtTm'], from_pattern2).cast("timestamp")) \
  .drop('HospitalDtTm') \
  .withColumn('AvailableDtTmTS', unix_timestamp(fireServiceCallsDF['AvailableDtTm'], from_pattern2).cast("timestamp")) \
  .drop('AvailableDtTm')  
fireServiceCallsTsDF.printSchema()
# Notez comme l'affichage des dates/timestamps est différente
fireServiceCallsTsDF.take(5)<jupyter_output><empty_output><jupyter_text>Maintenant, trouvez la liste d'années qui sont enregistrées dans le CSV.Vous pouvez utiliser la fonction sql `year()` pour vous aider : http://spark.apache.org/docs/2.1.0/api/python/pyspark.sql.htmlpyspark.sql.functions.year<jupyter_code># votre code ici<jupyter_output><empty_output><jupyter_text>Question 4 : Combien d’appels ont été enregistrés lors des 7 derniers jours ?ATTENTION : Considérez que nous sommes le 15 juin 2003 (le 166ème jour de l'année). Utilisez la date sur le colonne `CallDateTS`.Vous pouvez filtrer les entrées avec les deux fonctions utilitaires SQL `year()` et `dayofyear()`.<jupyter_code># votre code ici<jupyter_output><empty_output><jupyter_text>Note à propos de la performanceSi vous avez fait attention, toutes les opérations précédentes prendent un certain temps à s'exécuter. L'une des raisons est le fait que à chaque fois on oblige la lecture du fichier. Il serait plus rapide si on pouvait garder ces données en cache mémoire, pour un accès plus rapide.<jupyter_code># ici on va définir une view qui sera mise en cache lors de l'exécution d'une action
fireServiceCallsTsDF.createOrReplaceTempView("fireServiceVIEW")
spark.catalog.cacheTable("fireServiceVIEW")
# on fait l'action count() pour mettre la table en mémoire.  
spark.table("fireServiceVIEW").count()
fireServiceDF = spark.table("fireServiceVIEW")
# Le count() fait sur un DF en mémoire cache est bien plus rapide !!!
fireServiceDF.count()<jupyter_output><empty_output><jupyter_text>Jusqu'à présent nous avons fait des requêtes utilisant la syntaxe des Dataframes. Spark supporte aussi une syntaxe SQL à partir de la bibliothèse Spark.SQL. Si on programme directement en python, on pourrait faire des appels comme ceci :<jupyter_code>sqlWay = spark.sql("SELECT count(*) FROM fireServiceVIEW")
sqlWay<jupyter_output><empty_output><jupyter_text>Question 5 : Quel quartier a généré plus d'appels en 2002 ?Utilisez `spark.sql` pour faire votre requête. Le quartier est indiqué dans la colonne `NeighborhoodDistrict`. Pour les requêtes, utilisez le dataset `fireServiceVIEW` qui est déjà en cache. Limitez la réponse aux 15 quartiers avec le plus grand nombre d'appels.<jupyter_code># votre code ici<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.10417718779308852, 'fraction_numerical': 0.006060168818988529, 'mean_word_length': 4.242813918305598, 'pattern_counts': {'":': 0, '<': 49, '<?xml version=': 0, '>': 49, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '12', 'dir': 'jupyter-structured-clean-dedup', 'id': '597485', 'license': 'BSD-3-Clause', 'n_tokens_mistral': 4300, 'n_tokens_neox': 4195, 'n_words': 1214, 'path': '.old/3-DataFrames.ipynb', 'repo_name': 'lsteffenel/pyspark-binder', 'size': '13861'}