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

# # NOAA : Extraction de la pluviométrie
#
# Dataset :
# - Dossier /demo/noaa/noaa-raw-data
# - Format du fichier : "STATION","DATE","LATITUDE","LONGITUDE","ELEVATION","NAME","PRCP","PRCP_ATTRIBUTES"
# - Une ligne par relevé météo par jour
#
# Exercice :
# - Récupérer le code pays sur 2 lettres et la placer dans une nouvelle colonne "COUNTRY"
# - Supprimer les lignes où PRCP est null
# - Regrouper les relevés des stations par pays
# - Calculer la moyenne de pluie ("PRCP"), la somme, le maximum, l'écart type, le nombre de stations météo
# - Trier par date et par pays
# - Sauver dans '/stagiaire/votre_dossier/noaa/daily_rain_by_country'

# %%writefile spark_noaa.py
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SparkSession
import pyspark.sql.functions as F
spark = SparkSession \
    .builder \
    .appName('Spark RDD') \
    .getOrCreate()
sc = spark.sparkContext
lines = spark.read.text('/demo/noaa/noaa-raw-data/FRW00034067.csv') 
def attributes(ligne):
    station_name = ligne[0].replace('"', '')
    return [station_name] + ligne[7].split(',')
mon_rdd=lines.rdd.map(lambda r: r[0].split('","')).map(attributes).filter(lambda x: x[1]!='PRCP_ATTRIBUTES')
mon_rdd.take(2)
sc.stop()



# +
# #!hdfs dfs -ls /demo/noaa/noaa-raw-data
# -

from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SparkSession
import pyspark.sql.functions as F

# +
DATASET_FOLDER = '/demo/noaa/noaa-raw-data'
SPARK_MASTER = 'spark://localhost:7077'
APP_NAME = 'Group daily rain by Country - 430'

noaa_csv_path = DATASET_FOLDER 
output = '/stagiaire/votre_dossier/noaa/daily_rain_by_country'

# Create Spark session
spark = SparkSession.builder.master(SPARK_MASTER).appName(APP_NAME).getOrCreate()
# -

#
#     Récupérer le code pays sur 2 lettres et la placer dans une nouvelle colonne "COUNTRY"
#     Supprimer les lignes où PRCP est null
#     Regrouper les relevés des stations par pays
#     Calculer la moyenne de pluie ("PRCP"), la somme, le maximum, l'écart type, le nombre de stations météo
#     Trier par date et par pays
#     Sauver dans '/stagiaire/votre_dossier/noaa/daily_rain_by_country'
#

# +
# Utiliser RDD pour lire le fichier et extraire le code pays, la date et la température
# -



# +
#lines = spark.read.text('/demo/noaa/noaa-raw-data/FRW00034067.csv') 
#lines = spark.read.text('/demo/noaa/noaa-raw-data/FRE00104040.csv') 
# -

lines.rdd.map(lambda r: r).take(3)

lines.rdd.map(lambda r: r).take(3)

lines.rdd.map(lambda r: r[0]).take(3)

lines.rdd.map(lambda r: r[0][:5] ).take(3)

lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).take(3)

lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).map(lambda x: x[0] + x[1]).take(3)

lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).map(lambda x: x[0] + x[1]).count()

lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).map(lambda x: x[0] + x[1]).flatMap(lambda r: r).count()

lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).map(lambda x: x[0] + x[1]).flatMap(lambda r: r).take(3)



lines.rdd.map(lambda r: r[0].split(',')).take(1)

lines.rdd.map(lambda r: r[0].split(',')).count()

lines.rdd.map(lambda r: r[0].split(',')).flatMap(lambda r: r).take(1)

lines.rdd.map(lambda r: r[0].split(',')).flatMap(lambda r: r).count()

lines.rdd.map(lambda r: r[0].split(',')).map(lambda x: [x[0].replace('"',''), x[3].replace('"','')]).take(5)

# # Exercice
# Récupérer la colonne PRCP_ATTRIBUTES et la découper pour obtenir 3 colonnes

mon_rdd = lines.rdd.map(lambda r: r[0].split('","'))
mon_rdd.take(1)

mon_rdd=lines.rdd.map(lambda r: r[0].split('","')).map(lambda x: x[7].split(','))
mon_rdd.take(2)

['a'] + ['b']

mon_rdd=lines.rdd.map(lambda r: r[0].split('","')).map(lambda x: x[7].split(',')).filter(lambda x: x[0]!='PRCP_ATTRIBUTES')
mon_rdd.take(2)


def attributes(ligne):
    station_name = ligne[0].replace('"', '')
    return [station_name] + ligne[7].split(',')


mon_rdd=lines.rdd.map(lambda r: r[0].split('","')).map(attributes).filter(lambda x: x[1]!='PRCP_ATTRIBUTES')
mon_rdd.take(2)

df = spark.createDataFrame(mon_rdd).toDF('STATION','col1', 'col2', 'col3')
df.show(3)







lines.rdd.map(lambda r: (r[0][:5],r[0][5:10]) ).flatMap(lambda r: r).take(10)

lines.rdd.flatMap(lambda r: r).take(10)



df.write.csv(output, header=True)
print(df.columns)

sc.stop()



StarcoderdataJupyterScriptsDedupFiltered