<jupyter_start><jupyter_text>Working with NOAA datasetThe 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<jupyter_code>#!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
from pyspark.sql import SparkSession
import shutil
# import os

# os.environ["PYSPARK_DRIVER_PYTHON"] = "/usr/bin/python3"
# os.environ["PYSPARK_PYTHON"]="/usr/bin/python3"<jupyter_output><empty_output><jupyter_text>Line count<jupyter_code>#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 + '/france/*.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}')<jupyter_output>624,444<jupyter_text>Number of lines in the full dataset : 1 076 166 433 Filtershttps://docs.opendata.aws/noaa-ghcn-pds/readme.htmlhttps://www1.ncdc.noaa.gov/pub/data/ghcn/daily/readme.txtKeep only ELEMENT SummaryThe 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 showerWMO 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<jupyter_code>!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
!head -3 /media/data-nvme/dev/datasets/WorldBank/france/FRM00007180.csv
!rm -r /media/data-nvme/dev/datasets/WorldBank/year_2016-01
s ='"FRM00007180","1944-11-03","48.692","6.23","228.9","ESSEY, FR",,,,,,,,,"   49","H,,S"'
s[15:20]<jupyter_output><empty_output><jupyter_text>Filter on Rain - Join version113 074 files with rain data, no need to filter before other operation.<jupyter_code>%%time
from pyspark import SparkContext
from pyspark import SparkConf
from pyspark.sql import SparkSession
import pyspark.sql.functions as F

import shutil

DATASET_FOLDER = '/media/data-nvme/dev/datasets/WorldBank/'
SPARK_MASTER = 'spark://192.168.0.9:7077'
APP_NAME = 'Group daily rain by Country'

noaa_csv_path = DATASET_FOLDER + '/noaa/*.csv'
noaa_csv_path = DATASET_FOLDER + 'noaa/ASN*.csv'
output = DATASET_FOLDER + 'daily_rain_by_country'

# Create Spark session
spark = SparkSession.builder.master(SPARK_MASTER).appName(APP_NAME).getOrCreate()
# Convert list to data frame
df = spark.read.format('csv').option('header',True).option('multiLine', True).load(noaa_csv_path)
df_rain = df[["STATION","DATE","LATITUDE","LONGITUDE","ELEVATION","NAME","PRCP","PRCP_ATTRIBUTES"]]

# Extract country code
df_rain = df_rain.withColumn('COUNTRY',  F.col('NAME').substr(F.length('NAME')-1, F.length('NAME')))

# Register the DataFrame as a SQL temporary view
df_rain.createOrReplaceTempView("noaa")
sqlDF = spark.sql("SELECT DATE, COUNTRY, ceil(100 * avg(PRCP))/100 as avg_PRCP, ceil(100 * sum(PRCP))/100 as sum_PRCP, max(PRCP) as max_PRCP, ceil(100 * stddev(PRCP))/100 as stddev_PRCP, count(PRCP) as count_PRCP FROM noaa GROUP BY DATE, COUNTRY;")

shutil.rmtree(output, ignore_errors=True)
#sqlDF.repartition(1).write.csv(output)
sqlDF.write.csv(output)
print(sqlDF.columns)
!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-fra'
noaa_csv_path = DATASET_FOLDER + '/france/*.csv'
inventory_path = DATASET_FOLDER + 'ghcnd-inventory.txt'
output = DATASET_FOLDER + 'france_rain'
#sc.stop()
total = 0
# configuration
APP_NAME = 'Filter on Rain'

# Create Spark session
spark = SparkSession.builder.master(SPARK_MASTER).appName(APP_NAME).getOrCreate()
# Convert list to data frame
df = spark.read.format('csv').option('header',True).option('multiLine', True).load(noaa_csv_path)

'","'.join(df.columns)
df_rain = df[["STATION","DATE","LATITUDE","LONGITUDE","ELEVATION","NAME","PRCP","PRCP_ATTRIBUTES"]]
#df_rain["COUNTRY"] = df_rain["NAME"][:-2]
#from pyspark.sql.functions import col
import pyspark.sql.functions as F

# Extract country code
df_rain = df_rain.withColumn('COUNTRY',  F.col('NAME').substr(F.length('NAME')-1, F.length('NAME')))
df_rain.head(3)
output = DATASET_FOLDER + 'france_rain'
df_rain.repartition(1).write.csv(output)
#df_rain.write.csv(output)
df_rain.columns
import pandas as pd
#df_out = pd.read_csv('/media/data-nvme/dev/datasets/WorldBank/daily_rain_by_country.csv.gz')
df_out = pd.read_csv('/media/data-nvme/dev/src/batch8_worldbank/datasets/daily_rain_by_country.csv.gz')
df_out.head(3)
df_out.tail(10)
df_out.columns
df_out.columns = ['date', 'country_ISO2', 'avg_rain', 'sum_rain', 'max_rain', 'stddev_rain', 'station_count']
df_out.columns
df_out = df_out.dropna()
df_out.to_csv('/media/data-nvme/dev/src/batch8_worldbank/datasets/daily_rain_by_country.csv.gz', compression='gzip', index=False)
import pandas as pd
df_out = pd.read_csv('/media/data-nvme/dev/datasets/WorldBank/france-rain.csv')
df_out.head(3)
df_test = df_out.query('DATE == "2016-04-01"')
df_test.describe()
# Register the DataFrame as a SQL temporary view
df_rain.createOrReplaceTempView("noaa")
sqlDF = spark.sql("SELECT DATE, COUNTRY, ceil(100 * avg(PRCP))/100 as avg_PRCP, ceil(100 * sum(PRCP))/100 as sum_PRCP, max(PRCP) as max_PRCP, ceil(100 * stddev(PRCP))/100 as stddev_PRCP, count(PRCP) as count_PRCP FROM noaa GROUP BY DATE, COUNTRY;")
sqlDF.head(3)
output = DATASET_FOLDER + 'france_rain_avg_by_day'
shutil.rmtree(output)
sqlDF.repartition(1).write.csv(output)
#df_rain.write.csv(output)
sqlDF.columns

','.join(sqlDF.columns)

sc = SparkContext(conf=conf)


full_df = sc.read.option("header", True).option("inferSchema", True).csv(noaa_csv_path)

# 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

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<jupyter_output><empty_output><jupyter_text>Filter on Year<jupyter_code>%%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'
noaa_csv_path = DATASET_FOLDER + '/france/*.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 : s[15:20] == "2016-" in s)
        output = DATASET_FOLDER + 'year_2016-fra'
        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

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[1:12]
s[14:]
s = 'ACW00011604  17.1167  -61.7833 TMAX 1949 1949'
s[0:11]
s[36:]<jupyter_output><empty_output><jupyter_text>https://stackoverflow.com/questions/56957589/how-to-read-multiple-csv-files-with-different-schema-in-pyspark<jupyter_code>!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()<jupyter_output><empty_output><jupyter_text>"ORLY" ID : 0-20000-0-07149Coordinates: 48.7166666667°N, 2.3844444444°E, 89m<jupyter_code># !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())<jupyter_output><empty_output><jupyter_text>TEST<jupyter_code>conf = SparkConf().setAppName('test')
conf = conf.setMaster(SPARK_MASTER)
sc = SparkContext(conf=conf)
s = ['"USR0000AFRA","2016-01-25","35.8456","-113.055","2063.5","<NAME>RIZONA, 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()<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.13626595879155606, 'fraction_numerical': 0.03994438124130957, 'mean_word_length': 4.142346441338966, 'pattern_counts': {'":': 0, '<': 38, '<?xml version=': 0, '>': 39, 'https://': 3, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 2, 'substrings_counts': 2, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '8', 'dir': 'jupyter-structured-clean-dedup', 'id': '535744', 'license': 'MIT', 'n_tokens_mistral': 6094, 'n_tokens_neox': 5604, 'n_words': 1276, 'path': 'model_innondations/notebook/spark_pipeline_01_noaa_spark.ipynb', 'repo_name': 'alencon/batch8_worldbank', 'size': '15852'}