<jupyter_start><jupyter_text>Export for Dashboard<jupyter_code># For multiple output per cell
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"
#DATASET_FOLDER = '/media/data-nvme/dev/datasets/WorldBank/'
DATASET_FOLDER = '../../datasets/precipitation/'

SPARK_MASTER = 'spark://192.168.0.9:7077'
APP_NAME = 'Merge PRCP GKP'
input_folder = DATASET_FOLDER
output = DATASET_FOLDER + '../wb_gkp_precipitation'
import os
import pandas as pd
import pandas as pd
from pyspark import SparkContext
#from pyspark import SparkConf
from pyspark.sql import SparkSession
#from pyspark.sql.window import Window
from pyspark.sql.types import FloatType
import pyspark.sql.functions as F

import shutil
### Connect to Spark
print('Create Spark session')
spark = SparkSession.builder.master(SPARK_MASTER).appName(APP_NAME).getOrCreate()
sc = spark.sparkContext<jupyter_output>Create Spark session<jupyter_text>History  Load EMDAT<jupyter_code>#!ls /media/data-nvme/dev/datasets/WorldBank/
# Load EMDAT
emdat = spark.read.format('csv').option('header',True).option('multiLine', True) \
    .load('/media/data-nvme/dev/datasets/WorldBank/' + 'emdat_public_2020_09_12_query_uid-tAnKEX-floods_only.csv')
print(emdat.columns)
#emdat = emdat.withColumn('decade',  F.concat(F.col('Year').substr(0, 3) , F.lit('0-') , F.col('Year').substr(0, 3), F.lit('9')))
emdat = emdat.withColumn('decade',  F.concat(F.col('Year').substr(0, 3) , F.lit('0')))
#emdat.take(3)
###### Cast Type
print('Cast string to float')
emdat = emdat.withColumn("Total Deaths", emdat["Total Deaths"].cast(FloatType()))\
    .withColumn("Total Damages ('000 US$)", emdat["Total Damages ('000 US$)"].cast(FloatType()))
emdat.createOrReplaceTempView("emdat")
emdat.count()
# Agregate by decade and country
emdat_agregate = spark.sql("""
SELECT decade as Decade, ISO, region as UN_Geosheme_Subregion, 'Flood' as Disaster_Type, 'past' as RCP,
    sum(`Total Damages ('000 US$)`) as Financial_Impact, sum(`Total Deaths`) as Human_Impact,
    count(`Dis No`) as DO
FROM emdat
GROUP BY decade, region, ISO
ORDER BY decade, region, ISO
""")

dfp = emdat_agregate.toPandas()
dfp.dropna().head(3)<jupyter_output><empty_output><jupyter_text>Load Rainfall<jupyter_code># Load Rainfall
DATASET_FOLDER = '../../datasets/WorldBank'
rain = spark.read.format('csv').option('header',True).option('multiLine', True) \
    .load(f'{DATASET_FOLDER}/daily_rain_by_country_feature.csv.gz')
# Create an agregate of rain by decade and country
rain.createOrReplaceTempView("noaa")
noaa = spark.sql("""
    SELECT decade, country_ISO3, count(avg_rain) as `nb_days_with_rain_>_50mm` FROM noaa
    WHERE avg_rain > 50
    GROUP BY decade, country_ISO3
    ORDER BY decade, country_ISO3
""")
noaa.createOrReplaceTempView("noaa_agregate")<jupyter_output><empty_output><jupyter_text>Join EMDAT and Rainfall<jupyter_code># Join EMDAT agregate and rain agregate
emdat_agregate.createOrReplaceTempView("emdat_agregate")
emdat_rain_agregate = spark.sql("""
    SELECT emdat_agregate.*, noaa_agregate.`nb_days_with_rain_>_50mm`
    FROM emdat_agregate LEFT JOIN noaa_agregate ON emdat_agregate.decade = noaa_agregate.decade
        AND emdat_agregate.ISO = noaa_agregate.country_ISO3
""")
emdat_rain_agregate.show(3)
# Agregate by region
emdat_rain_agregate.createOrReplaceTempView("emdat_rain_gregate")
emdat_rain_agregate_by_country = spark.sql("""
SELECT Decade, UN_Geosheme_Subregion, Disaster_Type, RCP, sum(Financial_Impact) as Financial_Impact, sum(Human_Impact) as Human_Impact, sum(DO) as DO, sum(`nb_days_with_rain_>_50mm`) as `nb_days_with_rain_>_50mm`
FROM emdat_rain_gregate
GROUP BY Decade, UN_Geosheme_Subregion, Disaster_Type, RCP
ORDER BY Decade, UN_Geosheme_Subregion
""")
# Filling NA/null with 0
emdat_rain_agregate_by_country = emdat_rain_agregate_by_country.fillna({'Financial_Impact' : 0, 'Human_Impact' : 0, 'nb_days_with_rain_>_50mm': 0})
emdat_rain_agregate_by_country.show(3)
#dfp = emdat.toPandas()
#dfp[['Dis No', "Total Deaths"]]
#dfp[dfp['Dis No'] == '1906-0023-BEL'] #") #1906-0023-BEL 	1950-0007-CHN
# Check that wr have good data
emdat_rain_agregate_by_country.createOrReplaceTempView("emdat_rain_agregate_by_country")

spark.sql("""
SELECT *
FROM emdat_rain_agregate_by_country
WHERE Financial_Impact > 0
ORDER BY Financial_Impact
""").show(3)
# Saving in one file with Pandas
dfp = emdat_rain_agregate_by_country.toPandas()
dfp
dfp.to_csv(f'{DATASET_FOLDER}../flood_history_agregates.csv.gz', index=False ,compression='gzip')<jupyter_output><empty_output><jupyter_text>Projection<jupyter_code># rain = spark.read.format('csv').option('header',True).option('multiLine', True) \
#     .load(f'{DATASET_FOLDER}../projection_preciptation_monthly_merged-2020-12-02')
sc.stop()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13268069152259945, 'fraction_numerical': 0.019995834201208082, 'mean_word_length': 6.950331125827814, 'pattern_counts': {'":': 0, '<': 20, '<?xml version=': 0, '>': 27, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 1, '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': '5', 'dir': 'jupyter-structured-clean-dedup', 'id': '19895', 'license': 'MIT', 'n_tokens_mistral': 1920, 'n_tokens_neox': 1811, 'n_words': 334, 'path': 'model_innondations/notebook/spark_pipeline_07_export_to_dashboard.ipynb', 'repo_name': 'alencon/batch8_worldbank', 'size': '4801'}