# ---
# jupyter:
#   jupytext:
#     text_representation:
#       extension: .py
#       format_name: light
#       format_version: '1.5'
#       jupytext_version: 1.14.4
#   kernelspec:
#     display_name: Python 3 (ipykernel)
#     language: python
#     name: python3
# ---

# ### Pour aller plus loin avec le SQL distribué
#
# Nous avons vu précédemment que l'on pouvait via Spark SQL accéder à des données distribuées stockées sous forme de fichier en les rêquetant avec du SQL standard.
#
# Pour cela, Spark traduit la requête SQL à éxecuter dans un plan d'éxécution et bien souvent :
# 1. Ramène sur les executeurs Spark les données du fichier.
# 2. Définit les traitements qu'il va faire sur les RDD pour arriver aux résultats.
# 3. Ramène le résultat sur le client, dans notre cas Jupyter.
#
# Pour avoir de bonnes performances, et d'autant plus que les executeurs Spark et les données ne sont pas sur les mêmes machines, il est donc utile de :
# 1. Avoir des données compressées qui transitent plus vite sur le réseau.
# 2. Avoir des mécanismes intelligents qui fitrent les données pour ne ramener sur les executeurs que les données réellement nécessaires.
#
# Pour cela, il existe différents formats de fichier, et si csv est l'un des plus répandus, ce n'est pas forcément le plus optimal.
# Nous allons ici vous présenter le format parquet qui est un format compressé, orienté colonne, avec des mécanismes de filtrage qui optimisent le transit de l'information réellement utile sur les executeurs.

# +
import os
from pyspark.sql import SparkSession

spark = (SparkSession 
         .builder
         # url par défaut d'une api kubernetes accédé depuis l'intérieur du cluster 
         # (ici le notebook tourne lui même dans kubernetes)
         .master("k8s://https://kubernetes.default.svc:443")
         # Nom du pod qui exécute le driver
         .config("spark.kubernetes.driver.pod.name", os.environ['KUBERNETES_POD_NAME'])
         # image des executors spark: pour des raisons de simplicité on réutilise l'image du notebook
         .config("spark.kubernetes.container.image", os.environ['IMAGE_NAME'])
         # Nom du namespace kubernetes
         .config("spark.kubernetes.namespace", os.environ['KUBERNETES_NAMESPACE'])
         # Nombre d'executeur spark, il se lancera autant de pods kubernetes que le nombre indiqué.
         .config("spark.executor.instances", "5")
         # Mémoire alloué à la JVM
         # Attention par défaut le pod kubernetes aura une limite supérieur qui dépend d'autres paramètres.
         # On manipulera plus bas pour vérifier la limite de mémoire totale d'un executeur
         .config("spark.executor.memory", "4g")
         .getOrCreate()
        )

sc = spark.sparkContext

# +
import os
import s3fs
import json

from pyspark.sql.types import StructType

endpoint = "https://"+os.environ['AWS_S3_ENDPOINT']
fs = s3fs.S3FileSystem(client_kwargs={'endpoint_url': endpoint})

with fs.open('s3://projet-spark-lab/diffusion/formation/schema/sirene/sirene.schema.json') as f:
    a = f.read()
    schema = StructType.fromJson(json.loads(a))

df = (spark.read
      .format("csv")
      .options(header='true', inferschema='false', delimiter=',')
      .schema(schema)
      .load("s3a://projet-spark-lab/diffusion/formation/data/sirene/sirene.csv")
     )
df.printSchema()
# -

# #### Requete sur le csv
#
# Nous allons faire 3 requetes sur cette table appelée sirene.
#
# 1. Un count du nombre de lignes
# 2. Un extrait de 10 lignes
# 3. Un comptages du top 10 des code activitePrincipalEtablissement.
#
# On va afficher le temps d'execution.

# +
df.createOrReplaceTempView("sirene")

dfCount = spark.sql("SELECT * FROM sirene LIMIT 10")
dfGroupBy= spark.sql("SELECT count(*) as tot , activitePrincipaleEtablissement FROM sirene group by activitePrincipaleEtablissement order by tot desc LIMIT 10")
# %time print('comptage total : {}'.format(df.count()))
# %time print('comptage des 10 premières lignes : {}'.format(dfCount.count()))
# %time dfGroupBy.show()
# -

# #### Le plan suivi par spark pour le groupby
#
# 1. Spark a lu tout le fichier en 53 étapes pour chaque étape il a calculé le nombre d'établissement par codeApe.
# 2. Il a mis ce résultat en commun pour aggréger
# 3. Il a ramené sur le client le top10
#
# Le shuffle indique généralement des échanges sur le réseau entre executor spark, plus il y a de shuffle moins on est content généralement.
# Quand on ne peut l'éviter, on souhaite qu'il concerne le moins de donnée possible, ici, ca va, le shuffle concerne un code de nomenclature et un total.
#
# (les temps d'execution ne sont peut etre pas les memes selon la configuration de votre spark)
#
# ![image.png](attachment:image.png)

# #### Comparaison avec les mêmes données au format parquet
#
# Le dataframe issu du csv a été enregistré au format parquet avec la commande suivante
# ```
# df.write.mode('append').parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene.parquet")
# ```
#
# Nous allons avec le sqlContext de Spark refaire exactement la même chose que précédemment.
# 1. Demander de lire le fichier en découvrant seul le schéma.
# 2. Donner le résultat des 3 mêmes requetes.
#
# Remarquez qu'une fois le dataFrame lu, les syntaxes sont exactement les mêmes.
#
#

# +
#df.write.mode('append').parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene.parquet")
# -

# !mc ls s3/projet-spark-lab/diffusion/formation/data/sirene.parquet

# Vos pouvez remarquer que 53 fichiers ont écrit sur le stockage. Cela correspond exactement exactement au nombre de partition du dataframe original. En effet spark dans sa configuration par défaut travaille sur 128 Mo de données et a donc découper le job sur les 6,5 Go en 53 taches qui chacune va écrire l'output dans un nouveau fichier.

parquetDf = spark.read.parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene.parquet")
parquetDf.createOrReplaceTempView("sireneparquet")
parquetDfCount = spark.sql("SELECT * FROM sireneparquet LIMIT 10")
parquetDFGroupBy= spark.sql("SELECT count(*) as tot , activitePrincipaleEtablissement FROM sireneparquet group by activitePrincipaleEtablissement order by tot desc LIMIT 10")
# %time print('comptage total : {}'.format(parquetDf.count()))
# %time print('comptage des 10 premières lignes : {}'.format(parquetDfCount.count()))
# %time parquetDFGroupBy.show()

# #### Le plan suivi par spark pour le groupby
#
# En parquet, pour le même group by il y a un gain important de performance, l'essentiel du temps est gagné du fait que le format est optimisé pour ne faire passer sur le réseau vers les executors que les données de la colonne concernée.
# On passe de 6Go de données brassées à quelques Mo.
#
# De plus le fichier parquet logique a été écrit en 53 sous fichiers (hérités des 53 partitions du csv).
# Mais comme les fichiers sont compressés en parquet il n'y a que 9 blocs de lus au lieu de 53.
#
# Une fois la lecture faite des fichiers on retrouve la seconde étape avec 200 partitions, c'est la valeur par défaut dans spark que l'on retrouve dans l'execution csv quand il doit restructurer son rdd sur le réseau.
#
# ![image.png](attachment:image.png)

# #### amélioration possible
#
# Le dataframe issu du csv a été enregistré au format parquet en regardant la structure du fichier enregistré on 
# on a hérité ca lors de l'écriture.
#
# Parquet ou non, on peut forcer l'écriture dans un fichier  en réorganisant le partitionnement du rdd avant l'écriture :
# On peut par exemple chercher à avoir des fichiers de 128 Mo environ pour lire un seul fichier par tache.
# La tache de repartition est assez couteuse et nous sommes obligés d'augmentant les limites de mémoire du container pour éviter les erreurs.
# ```
# config("spark.kubernetes.memoryOverheadFactor", "0.5")
# parquetDf.repartition(10).write.mode('overwrite').parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene-10-partitions.parquet")
# ```
#
# Tant qu'on y est changeons le niveau de parallélisme lors du shuffle de spark qui est par défaut à 200.
# ```
# config("spark.sql.shuffle.partitions", "5")
# config("spark.default.parallelism", "5")
# ```
#

spark.stop()
spark = (SparkSession 
         .builder
         # url par défaut d'une api kubernetes accédé depuis l'intérieur du cluster 
         # (ici le notebook tourne lui même dans kubernetes)
         .master("k8s://https://kubernetes.default.svc:443")
         # Nom du pod qui exécute le driver
         .config("spark.kubernetes.driver.pod.name", os.environ['KUBERNETES_POD_NAME'])
         # image des executors spark: pour des raisons de simplicité on réutilise l'image du notebook
         .config("spark.kubernetes.container.image", os.environ['IMAGE_NAME'])
         # Nom du namespace kubernetes
         .config("spark.kubernetes.namespace", os.environ['KUBERNETES_NAMESPACE'])
         # Nombre d'executeur spark, il se lancera autant de pods kubernetes que le nombre indiqué.
         .config("spark.executor.instances", "5")
         .config("spark.sql.shuffle.partitions", "5")
         .config("spark.default.parallelism", "5")
         .config("spark.kubernetes.memoryOverheadFactor", "0.5")
         # Mémoire alloué à la JVM
         # Attention par défaut le pod kubernetes aura une limite supérieur qui dépend d'autres paramètres.
         # On manipulera plus bas pour vérifier la limite de mémoire totale d'un executeur
         .config("spark.executor.memory", "4g")
         .getOrCreate()
        )

parquetDf = spark.read.parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene.parquet")

# +
#parquetDf.repartition(10).write.mode('overwrite').parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene-10-partitions.parquet")
# -

# !mc ls s3/projet-spark-lab/diffusion/formation/data/sirene-10-partitions.parquet

tenParquetDf = spark.read.parquet("s3a://projet-spark-lab/diffusion/formation/data/sirene-10-partitions.parquet")
tenParquetDf.createOrReplaceTempView("sirene10")
tenParquetDfCount = spark.sql("SELECT * FROM sirene10 LIMIT 10")
tenParquetDfGroupBy= spark.sql("SELECT count(*) as tot , activitePrincipaleEtablissement FROM sirene10 group by activitePrincipaleEtablissement order by tot desc LIMIT 10")
# %time print('comptage total : {}'.format(tenParquetDf.count()))
# %time print('comptage des 10 premières lignes : {}'.format(tenParquetDfCount.count()))
# %time tenParquetDfGroupBy.show()

# Normalement il y a, selon le use case, des paramètres à trouver :
# * nombre de coeurs disponibles (une tache = un coeur)
# * volume des données dans chaque étape (pas d'intéret d'avoir plein de petites étapes sur petites données)
# * nombre de fois ou le traitement ne peut faire autre chose que shuffle
#

parqDFGroupBy= spark.sql(
    "select count(*) as total, activitePrincipaleEtablissement, word from ("
    " SELECT activitePrincipaleEtablissement,explode(split(denominationUsuelleEtablissement,' ')) as word "+
    " FROM sirene10 where denominationUsuelleEtablissement is not null)"+
    " where length(word)>4"+
    " group by activitePrincipaleEtablissement,word order by total desc")
# %time parqDFGroupBy.show()

# #### Partitionnement avec parquet
#
# Si votre use case suppose que vous allez souvent requeter avec une clause where filtrante comme une date, une notion de géographie par exemple ou autre, on peut avec parquet partitionner sur plusieurs niveaux.
#
# Imaginons qu'un use case nécessite que les données soient préparées pour etre requetes par codeApe.
#
# La table a été préparé et partitionné par codeApe.
# - un codeApe = 1 fichier parquet
#
# Sur du gros volume cela génère un gain.
# Sur du petit volume cela peut etre une perte ici par exemple génère plus de 2000 fichiers ape qui va donc a la lecture commencer par 2000 petites step spark ce qui va etre long.
#
# L'idéal est donc d'avoir un use case ou une partition > bloc standard s3.
#
# ```tenParquetDf.write.partitionBy("etatAdministratifEtablissement").parquet("s3a://projet-spark-lab/diffusion/formation/data/sireneEae.parquet")```

# +
#tenParquetDf.write.partitionBy("etatAdministratifEtablissement").parquet("s3a://projet-spark-lab/diffusion/formation/data/sireneEae.parquet")
# -

# !mc ls s3/projet-spark-lab/diffusion/formation/data/sireneEae.parquet

# !mc ls s3/projet-spark-lab/diffusion/formation/data/sireneEae.parquet/etatAdministratifEtablissement=A/

tenParquetDf.groupBy('etatAdministratifEtablissement').count().show()

# On peut retenir cette variable ici pour l'exemple car elle partitionne les données sans trop les fragmenter.
#
# En effet, le partitionement va splitter le fichier parquet en autant de fichier parquet que de modalité donc :
# * si l'on prend une variable trop discriminante, on va pénaliser les requetes qui s'executeront sur tout le dataset (exemple si on partitionne sur codeApe avec plus de 2000 modalités => 2000 fichiers parquet a parcourir pour traiter tout le dataset donc spark commencera par une étape avec 2000 taches une pour chaque fichier ce qui va ralentir le traitement il y aura trop d'overhead).
# * Le partitionnement s'applique donc sur des variables cohérentes par rapport aux blocs de données configurés sur s3 et aux volumes de données naturellement distribuées dans la partition choisie.
# * Sauf a etre sur que toutes les requetes présenteront un filtre portant sur la partition ca peut avoir de grandes incidences sur le dataset global.
#
# On a donc 3 fichiers en dessous de notre parquet un par modalité 
#
# ![image.png](attachment:image.png)
#
# En théorie on doit : 
# * gagner du temps si l'on utilise la varible etatAdministratifEtablissement.
# * ne pas avoir trop dégrader les perfs sur le dataset global puisque la partition n'a pas fait explosé le nombre de fichier parquet par rapport au bloc s3 de 180Mo.

#on doit gagner du temps sur une partition
eaeDF = spark.read.parquet("s3a://projet-spark-lab/diffusion/formation/data/sireneEae.parquet")
eaeDF.createOrReplaceTempView("sirenepartition")
parqDFGroupBy= spark.sql(
    " select count(*) as total,activitePrincipaleEtablissement, word from ("
    " SELECT activitePrincipaleEtablissement,explode(split(denominationUsuelleEtablissement,' ')) as word "+
    " FROM sirenepartition where denominationUsuelleEtablissement is not null  and etatAdministratifEtablissement='A') "+
    " where length(word)>4"+
    " group by activitePrincipaleEtablissement,word order by total desc")
# %time parqDFGroupBy.show()

#on doit pas avoir dégrader le traitement sur le dataset global
parqDFGroupBy= spark.sql(
    " select count(*) as total,activitePrincipaleEtablissement, word from ("
    " SELECT activitePrincipaleEtablissement,explode(split(denominationUsuelleEtablissement,' ')) as word "+
    " FROM sirenepartition where denominationUsuelleEtablissement is not null) "+
    " where length(word)>4"+
    " group by activitePrincipaleEtablissement,word order by total desc")
# %time parqDFGroupBy.show()

# ### Synthèse
#
# Nous avons pu voir que les temps de traitements dépendent de :
# 1. Comment la donnée est enregistrée(format, granularité des blocs, partitionnement)
# 2. De paramètres de parallélisme des taches.
# 3. La capacité à l'utilisateur de comprendre ce qui se passe côté serveur et côté client pour ne ramener sur le client que le résultat.
#
# Dans ce tutoriel, vous maitrisez votre cluster et votre fichier dans une stratégie de datalake d'entreprise:
# - la diffusion des tables doit être réfléchie par une alimentation des tables/pipeline dans un format approprié au context du contenu.
# - l'utilisateur accède aux tables sans en connaitre précisement le stockage.

spark.stop()



StarcoderdataJupyterScriptsDedupFiltered