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

# # Working with NOAA dataset
#
# The journal article describing GHCN-Daily is:
#
# <NAME>., <NAME>, <NAME>, <NAME>, and <NAME>, 2012: An overview of the Global Historical Climatology Network-Daily Database. Journal of Atmospheric and Oceanic Technology, 29, 897-910, doi:10.1175/JTECH-D-11-00103.1

# +
# #!pip install pyspark
# -

# For multiple output per cell
from IPython.core.interactiveshell import InteractiveShell
InteractiveShell.ast_node_interactivity = "all"

DATASET_FOLDER = '/media/data-nvme/dev/datasets/WorldBank/'
noaa_csv_path = DATASET_FOLDER + 'noaa/ASN*.csv'
SPARK_MASTER = 'spark://192.168.0.9:7077'

from pyspark import SparkContext
from pyspark import SparkConf
import shutil

# +
# import os

# os.environ["PYSPARK_DRIVER_PYTHON"] = "/usr/bin/python3"
# os.environ["PYSPARK_PYTHON"]="/usr/bin/python3"

# -

# # Compute Pi

# +
conf = SparkConf().setAppName('pi')
conf = conf.setMaster(SPARK_MASTER)
sc = SparkContext(conf=conf)
import random
num_samples = 100000000
def inside(p):     
  x, y = random.random(), random.random()
  return x*x + y*y < 1

count = sc.parallelize(range(0, num_samples)).filter(inside).count()

pi = 4 * count / num_samples
print(pi)

sc.stop()
# -

# # Line count

# +
#noaa_csv_path = '/media/data-nvme/dev/datasets/WorldBank//noaa/ASN00060066.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/ASN*.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + '/small_dataset/*.csv'
def count_lines():
    # configuration
    APP_NAME = 'count NOAA all lines'
    conf = SparkConf().setAppName(APP_NAME)
    conf = conf.setMaster(SPARK_MASTER)
    sc = SparkContext(conf=conf)
    # core part of the script
    files = sc.textFile(noaa_csv_path)
    total = files.count()
#     lineLength = lines.map(lambda s: len(s))
#     lineLength.persist()
#     totalLength = lineLength.reduce(lambda a,b:a+b)
#     # output results
#     print(totalLength)
    sc.stop()
    return total

total = count_lines()
#print(f'{totalLength:%.2f}')
total

# -

print(f'{total:,d}')

# Number of lines : 1 076 166 433
#
# # Filters
#
# https://docs.opendata.aws/noaa-ghcn-pds/readme.html
#
# https://www1.ncdc.noaa.gov/pub/data/ghcn/daily/readme.txt
#
# Keep only 
#
# ELEMENT Summary
#
# The five core elements are:
#
#     PRCP = Precipitation (tenths of mm)
#     SNOW = Snowfall (mm)
#     SNWD = Snow depth (mm)
#     TMAX = Maximum temperature (tenths of degrees C)
#     TMIN = Minimum temperature (tenths of degrees C)
#
#     Variable   Columns   Type
#     ------------------------------
#     ID            1-11   Character
#     YEAR         12-15   Integer
#     MONTH        16-17   Integer
#     ELEMENT      18-21   Character
#     VALUE1       22-26   Integer
#     MFLAG1       27-27   Character
#     QFLAG1       28-28   Character
# 		   
# VALUE1     is the value on the first day of the month (missing = -9999).
#
#       "PRCP_ATTRIBUTES" = a,M,Q,S where:
#            a = DaysMissing (Numeric value): The number of days (from 1 to 5) missing or flagged is provided   
#            M = GHCN-Daily Dataset Measurement Flag (see Section 1.3.a.ii for more details) 
#            Q = GHCN-Daily Dataset Quality Flag (see Section 1.3.a.iii for more details)
#            S = GHCN-Daily Dataset Source Code (see Section 1.3.a.iv for more details)  
#
# MFLAG1     is the measurement flag for the first day of the month.  There are
#            ten possible values:
#
#            Blank = no measurement information applicable
#            B     = precipitation total formed from two 12-hour totals
#            D     = precipitation total formed from four six-hour totals
# 	   H     = represents highest or lowest hourly temperature (TMAX or TMIN) 
# 	           or the average of hourly values (TAVG)
# 	   K     = converted from knots 
# 	   L     = temperature appears to be lagged with respect to reported
# 	           hour of observation 
#            O     = converted from oktas 
# 	   P     = identified as "missing presumed zero" in DSI 3200 and 3206
#            T     = trace of precipitation, snowfall, or snow depth
# 	   W     = converted from 16-point WBAN code (for wind direction)
#
# QFLAG1     is the quality flag for the first day of the month.  There are 
#            fourteen possible values:
#
#            Blank = did not fail any quality assurance check
#            D     = failed duplicate check
#            G     = failed gap check
#            I     = failed internal consistency check
#            K     = failed streak/frequent-value check
# 	   L     = failed check on length of multiday period 
#            M     = failed megaconsistency check
#            N     = failed naught check
#            O     = failed climatological outlier check
#            R     = failed lagged range check
#            S     = failed spatial consistency check
#            T     = failed temporal consistency check
#            W     = temperature too warm for snow
#            X     = failed bounds check
# 	   Z     = flagged as a result of an official Datzilla 
# 	           investigation
#     
# - WESF = Water equivalent of snowfall (tenths of mm)
#
# WV** = Weather in the Vicinity where ** has one of the following values:
#
#     01 = Fog, ice fog, or freezing fog (may include heavy fog)
#     03 = Thunder
#     07 = Ash, dust, sand, or other blowing obstruction
#     18 = Snow or ice crystals
#     20 = Rain or snow shower
#
# WMO ID is the World Meteorological Organization (WMO) number for the station. If the station has no WMO number (or one has not yet been matched to this station), then the field is blank.
#
#      
#
# HCN/CRN FLAG = flag that indicates whether the station is part of the U.S. Historical Climatology Network (HCN). There are three possible values:
#
#     Blank = Not a member of the U.S. Historical Climatology or U.S. Climate Reference Networks
#     HCN = U.S. Historical Climatology Network station
#     CRN = U.S. Climate Reference Network or U.S. Regional Climate Network Station
#

# !head /media/data-nvme/dev/datasets/WorldBank//noaa/ASN00060066.csv'

# !head $DATASET_FOLDER/noaa/ASN00060066.csv
# !wc -l $DATASET_FOLDER/noaa/ASN00060066.csv
# !grep 2016 $DATASET_FOLDER/noaa/AE000041196.csv | head -1

# !rm -r /media/data-nvme/dev/datasets/WorldBank/year_2016-01

# # Filter on Year

# +
# %%time
#noaa_csv_path = '/media/data-nvme/dev/datasets/WorldBank//noaa/ASN00060066.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/AE*.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + '/small_dataset/*.csv'

def filter_year():
    total = 0
    # configuration
    APP_NAME = 'Filter on Year'
    conf = SparkConf().setAppName(APP_NAME)
    conf = conf.setMaster(SPARK_MASTER)
    sc = SparkContext(conf=conf)
    try:
        # core part of the script
        print(f'Processing {noaa_csv_path}...')
        files_rdd = sc.textFile(noaa_csv_path)
        year_2016  = files_rdd.filter(lambda s : "\"2016-01-01" in s)
        output = DATASET_FOLDER + 'year_2016-01'
        print(f'Saving to {output}')
        year_2016.saveAsTextFile(output)
        total = year_2016.count()
    except Exception as inst:
        print('ERROR')
        print(type(inst))    # the exception instance
        print(inst.args)     # arguments stored in .args
        print(inst)
        raise
    finally:
        sc.stop()
        return total

total = filter_year()
#print(f'{totalLength:%.2f}')
total

# -

# # Filter on Rain - Python IN version

s = '"USR0000AFRA","2016-01-25","35.8456","-113.055","2063.5","<NAME>RIZONA, AZ US","   39","H,,U","  -39","H,,U","  -10",",,U"'
s[1:12]
s[14:]
s = 'ACW00011604  17.1167  -61.7833 TMAX 1949 1949'
s[0:11]
s[36:]

# +
# %%time
#noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + 'year_2016-01.csv'
inventory_path = DATASET_FOLDER + 'ghcnd-inventory.txt'
def extract_rain():
    total = 0
    # configuration
    APP_NAME = 'Filter on Rain'
    conf = SparkConf().setAppName(APP_NAME)
    conf = conf.setMaster(SPARK_MASTER)
    sc = SparkContext(conf=conf)
    try:
        # Load inventory
        inventory_rdd = sc.textFile(inventory_path)
        rain_inventory_rdd = inventory_rdd.filter(lambda s : "PRCP" in s)
        # Now we have a list of all files containings precipitation data
        # Keep only the first column
        rain_inventory_rdd = rain_inventory_rdd.filter(lambda s: s[0:11])
        rain_inventory = rain_inventory_rdd.collect()
        # Load the stations data points
        print(f'Processing {noaa_csv_path}...')
        files_rdd = sc.textFile(noaa_csv_path+'/*')
        # Keep only precipitation data
        rain_rdd  = files_rdd.filter(lambda s : s[1:12] in rain_inventory)
        # Save precipitation data
        rain_rdd.saveAsTextFile(DATASET_FOLDER + 'year_2016-01_rain')
        total = rain_rdd.count()
    except Exception as inst:
        print('ERROR')
        print(type(inst))    # the exception instance
        print(inst.args)     # arguments stored in .args
        print(inst)
        raise
    finally:
        sc.stop()
        return total

total = extract_rain()
#print(f'{totalLength:%.2f}')
total
# -

# # Filter on Rain - Join version

# !rm -r /media/data-nvme/dev/datasets/WorldBank/year_2016-01_rain

# +
# %%time
#noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + 'year_2016-01'
inventory_path = DATASET_FOLDER + 'ghcnd-inventory.txt'
sc.stop()
total = 0
# configuration
APP_NAME = 'Filter on Rain'
conf = SparkConf().setAppName(APP_NAME)
conf = conf.setMaster(SPARK_MASTER)
sc = SparkContext(conf=conf)

# Load inventory
inventory_rdd = sc.textFile(inventory_path)
# Filter on Rain
rain_inventory_rdd = inventory_rdd.filter(lambda s : "PRCP" in s)
# Now we have a list of all files containings precipitation data
# Keep only the first column
# Format each RDD as (K, V) to prepare for the join operation
rain_inventory_rdd = rain_inventory_rdd.map(lambda line : (line[0:11], line[36:])) # Keep code and years
#rain_inventory_rdd = rain_inventory_rdd.filter(lambda s: s[0:11])
#rain_inventory = rain_inventory_rdd.collect()
# Load the stations data points
print(f'Processing {noaa_csv_path}...')
all_data_rdd = sc.textFile(noaa_csv_path+'/*')
all_data_rdd = all_data_rdd.map(lambda line: (line[1:12], line[14:]))
# Keep only precipitation data
join = rain_inventory_rdd.join(all_data_rdd)
#rain_rdd  = files_rdd.filter(lambda s : s[1:12] in rain_inventory)
# Save precipitation data
output = DATASET_FOLDER + 'year_2016-01_rain'
print(f'Saving to {output}')
join_output = join.map(lambda x: ','.join([x[0],x[1][1]]))
#sc.parallelize(join_output.take(2)).collect()
# Flatten the result
#join = rdd.map(lambda x: ','.join([x[0],x[1][0],x[1][1]]))
# Get all partition on one node, to have one file (don't do it for huge dataset)
join_output = join_output.repartition(1)

shutil.rmtree(output, ignore_errors=True)
join_output.saveAsTextFile(output)
total = join.count()
sc.stop()

total

# -

# https://stackoverflow.com/questions/56957589/how-to-read-multiple-csv-files-with-different-schema-in-pyspark

# !echo '"STATION","DATE","LATITUDE","LONGITUDE","ELEVATION","NAME","PRCP","PRCP_ATTRIBUTES","SNWD","SNWD_ATTRIBUTES","TMAX","TMAX_ATTRIBUTES","TMIN","TMIN_ATTRIBUTES","TAVG","TAVG_ATTRIBUTES"'
# !head $DATASET_FOLDER/year_2016-01_rain/part-00000

sc.parallelize(join.take(2)).collect()





# "ORLY" ID : 0-20000-0-07149
#
# Coordinates: 48.7166666667°N, 2.3844444444°E, 89m 

# +
# # !cd $DATASET_FOLDER/year_2016-01_rain && (ls | xargs cat)  > ../year_2016-01_rain.csv
# # ! head $DATASET_FOLDER/year_2016-01_rain.csv

# +
#noaa_csv_path = '/media/data-nvme/dev/datasets/WorldBank//noaa/ASN00060066.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/AE*.csv'
#noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + 'year_2016-01.csv'
def extract_rain():
    total = 0
    # configuration
    APP_NAME = 'Count Rain'
    conf = SparkConf().setAppName(APP_NAME)
    conf = conf.setMaster(SPARK_MASTER)
    sc = SparkContext(conf=conf)
    try:
        # core part of the script
        print(f'Processing {noaa_csv_path}...')
        files_rdd = sc.textFile(noaa_csv_path+'/*')
        total = files_rdd.count()
    except Exception as inst:
        print('ERROR')
        print(type(inst))    # the exception instance
        print(inst.args)     # arguments stored in .args
        print(inst)
        raise
    finally:
        sc.stop()
        return total

total = extract_rain()
#print(f'{totalLength:%.2f}')
total
# -

# !head $DATASET_FOLDER/year_2016.csv

# +
from  pyspark.sql.functions import input_file_name
from pyspark.sql import SQLContext
from pyspark.sql.types import *
sqlContext = SQLContext(sc)

customSchema = StructType([ \
StructField("asset_id", StringType(), True), \
StructField("price_date", StringType(), True), \
etc., 
StructField("close_price", StringType(), True), \
StructField("filename", StringType(), True)])



df = spark.read.format("csv") \
   .option("header", "false") \
   .option("sep","|") \
   .schema(customSchema) \
   .load(fullPath) \
   .withColumn("filename", input_file_name())
# -

# # TEST

conf = SparkConf().setAppName('test')
conf = conf.setMaster(SPARK_MASTER)
sc = SparkContext(conf=conf)
s = ['"USR0000AFRA","2016-01-25","35.8456","-113.055","2063.5","FRAZIER WELLS ARIZONA, AZ US","   39","H,,U","  -39","H,,U","  -10",",,U"']
s += ['"FRM00007149","2016-01-01","48.7167","2.3842","89.0","ORLY, FR","   10",",,E",,,"   85",",,E","   38",",,E","   62","H,,S"']
s

rdd = sc.parallelize(s)
#test = rain_inventor_rdd.map(lambda line : (line[0:11], line[36:])) # Keep code and years
rdd = rdd.map(lambda line : (line[1:11], (line[15:25], line[28:35])))
rdd.collect()

rdd.repartition(1).collect()

rdd.map(lambda x: ','.join([x[0],x[1][0],x[1][1]])).repartition(1).collect()

sc.stop()



StarcoderdataJupyterScriptsDedupFiltered