<jupyter_start><jupyter_text>AWS + Spark: Exploration of NOAA Weather DataThis notebook demonstrates how to format and plot data with the *leaflet* class, which uses ipyleaflet for data visualization on an interactive map. Import libraries.The main utilities are the *leaflet* class which plots data and some functions for pulling data from the S3 bucket.The utilities library also contains other functions that the professor uses throughout the notebooks, but they are not necessarily used here<jupyter_code>import os.path as path
import sys
import numpy as np
from pyspark import SparkContext, SparkConf
from lib import *
from pyspark.sql import *
from lib.utils import *

import findspark
findspark.init()

%pylab inline<jupyter_output><empty_output><jupyter_text>Create new SparkContextSet path for the data directory (you will need to manually create ../DataHW3 if it doesn't already exist)<jupyter_code>!mkdir ../DataHW3
# Create a new spark and sql context
sc = create_sc(pyFiles=['lib/numpy_pack.py','lib/spark_PCA.py','lib/computeStatistics.py'])
sqlContext = SQLContext(sc)

# Create the names of the files/directories we're working with

data_dir = '../DataHW3'<jupyter_output><empty_output><jupyter_text>Check if we've already downloaded this fileIf we have not, we'll need to grab it from the course's S3 bucketHere are the states that we will look at: **Northern plains**: North Dakota, South Dakota, Minnesota, Iowa, Nebraska**Southern plains**: Texas, Oklahoma, Kansas<jupyter_code>data_dir = '../DataHW3'

if not path.exists(data_dir + '/' + 'stations.parquet'):
    getStations()

    
states = ['ND', 'SD', 'MN', 'IA', 'NE', 'TX', 'OK', 'KS']

for s in states:
    
    parquet = s + '.parquet'
    tarname = s + '.tgz'
    
    if not path.exists(data_dir + '/' + parquet):

        # pull the weather data for a particular state from the MAS-DSE S3 bucket
        getStateData(s, data_dir, tarname, parquet)<jupyter_output><empty_output><jupyter_text>Create a SQL context for our data... and perform some sample queries to see what our data looks like<jupyter_code># perform some sample queries to see what our data looks like

state = 'ND'
tarname = state + '.tgz'
parquet = state + '.parquet'
parquet_path = data_dir + '/' + parquet

stations_df = sqlContext.read.parquet(data_dir+'/stations.parquet')
sqlContext.registerDataFrameAsTable(stations_df, 'stations_table')


# create the sparksql context
df = sqlContext.read.parquet(parquet_path)
sqlContext.registerDataFrameAsTable(df,'table')




# Number of stations in ND that have measurement
Query="""
SELECT DISTINCT(Measurement)
FROM table
WHERE STATE=='ND'
"""
counts=sqlContext.sql(Query)
counts.show()

# First 5 measurements from the weather table. Note that it looks like 1 measurement is really = 1 year of measurements for that measurement type
Query = """
SELECT *
FROM table
LIMIT 5
"""
all_features = sqlContext.sql(Query)
all_features.show()

# Show the ND stations with the top number of measurements
Query = """
SELECT Station, Measurement, Values, longitude, latitude
FROM table
WHERE Measurement=='PRCP'
"""
weather_query = sqlContext.sql(Query)
weather_query.show()<jupyter_output><empty_output><jupyter_text>Below are the<jupyter_code># featureStr = "\'SNWD\'"

# dfs = []
# # create the sparksql context
# for s in states:
#     parquet = s + '.parquet'
#     parquet_path = data_dir + '/' + parquet
#     df = sqlContext.read.parquet(parquet_path)
#     sqlContext.registerDataFrameAsTable(df,f'table_{s}')
    
#     Query = f"""
#     SELECT Year, Station, state, Measurement, Values
#     FROM table_{s}
#     WHERE Measurement=={featureStr} and Year > 1969 and Year < 2010
#     """
#     weather_query = sqlContext.sql(Query)
#     rdd2 = weather_query.rdd.map(lambda x: replaceSNWD(x, 'Values'))
#     df2 = sqlContext.createDataFrame(rdd2).toPandas()
#     dfs.append(df2)

# all_states_SNWD = pd.concat(dfs)
# all_states_SNWD.to_excel('SNWD-All-States.xlsx')


# featureStr = "\'SNOW\'"

# dfs = []
# # create the sparksql context
# for s in states:
#     parquet = s + '.parquet'
#     parquet_path = data_dir + '/' + parquet
#     df = sqlContext.read.parquet(parquet_path)
#     sqlContext.registerDataFrameAsTable(df,f'table_{s}')
    
#     Query = f"""
#     SELECT Year, Station, state, Measurement, Values
#     FROM table_{s}
#     WHERE Measurement=={featureStr} and Year > 1969 and Year < 2010
#     """
#     weather_query = sqlContext.sql(Query)
#     rdd2 = weather_query.rdd.map(lambda x: replaceSNWD(x, 'Values'))
#     df2 = sqlContext.createDataFrame(rdd2).toPandas()
#     dfs.append(df2)

# all_states_SNOW = pd.concat(dfs)
# all_states_SNOW.to_excel('SNOW-All-States.xlsx')<jupyter_output><empty_output><jupyter_text>Here is an example of reducing the Values feature to a single value using a UDF (user defined function) from utilities<jupyter_code># This is the query that get's performed inside leaflet class, but Values have been cleaned
df = sqlContext.read.parquet(parquet_path)
sqlContext.registerDataFrameAsTable(df,'table')
Query = """
SELECT Station, Measurement, Values, longitude, latitude
FROM table
WHERE Measurement=='SNWD'
LIMIT 5
"""
weather_query = sqlContext.sql(Query)

rdd3 = weather_query.rdd.map(lambda x: replaceSNWD(x, 'Values'))
df2 = sqlContext.createDataFrame(rdd3)
df2.show()<jupyter_output><empty_output><jupyter_text>Data visualizationDepending on how many states' data we are plotting this could take a little while. The northern and souther plains combined took around ~5 min. Nevertheless, we can see the average snow depth during the first three calendar months of the year for these states. Note: I only included "ND" and "SD" for demonstration purposes, but it works with all of the statesFirst initialize the leaflet with sqlContext, and the feature represented as a string.Then proceed to add states data to the leaflet class by the .add(dataframe) method. **The dataframe should have the format of the query below**.Once everything has been added, you can run plot_all(). This is the step that will take a while for longer data setsTo show the map, run plotter.m, and to show the legend, run plotter.color_legend()<jupyter_code>featureStr = "\'SNWD\'"

plotter = leaflet_old(sqlContext, featureStr)

# create the sparksql context
for s in ["TX"]:
    parquet = s + '.parquet'
    parquet_path = data_dir + '/' + parquet
    df = sqlContext.read.parquet(parquet_path)
    sqlContext.registerDataFrameAsTable(df,f'table_{s}')
    
    Query = f"""
    SELECT Station, Measurement, Values, longitude, latitude
    FROM table_{s}
    WHERE Measurement=={featureStr}
    """
    weather_query = sqlContext.sql(Query)
    rdd2 = weather_query.rdd.map(lambda x: replaceSNWD(x, 'Values'))
    df2 = sqlContext.createDataFrame(rdd2)
    
    
    plotter.add(df2)
plotter.plot_all()
plotter.m
plotter.color_legend()
# class leaflet:
#     """
#     Plots circles on a map (one per station) whose size is proportional to the number of feature measurements, and whose color is equal to their aggregated value (min, max, avg)  
    
#     :param featureStr: the feature you'd like to visualize given as a string, eg. 'SNWD'
#     :param aggregateType: 'avg', 'min', 'max'
#     """
    
#     def __init__(self, sqlctxt, featureStr):
#         self.feat = featureStr
#         self.pdfMaster = None
#         self.sqlctxt = sqlctxt
        
#         self.maxLong = None
#         self.minLong = None
#         self.maxLat = None
#         self.minLat = None
        
#         self.minAggVal = None
#         self.maxAggVal = None
        
        
#         self.cmap = plt.get_cmap('jet') 
#         self.m = None
        
#     def add(self, dataframe):
#         """
#         Adds (or concatenates) a dataframe to the instance's master dataframe. The dataframe to be added must have atleast
#         the following columns: station, latitude, longitude, featureStr, value
#         """

#         self.sqlctxt.registerDataFrameAsTable(dataframe, "temp")
#         query = f"""
#         SELECT Station, latitude, longitude, COUNT(Measurement), MEAN(Values)
#         FROM temp
#         WHERE Measurement=='SNWD'
#         GROUP BY Station, latitude, longitude
#         """
#         tempdf = self.sqlctxt.sql(query).toPandas()
        
#         if (self.pdfMaster is None):
#             # set as new pdfMaster
#             self.pdfMaster = tempdf
            
#         else:            
#             # append to existing dfMaster
#             self.pdfMaster = pd.concat([self.pdfMaster, tempdf])
            
#     def plot_all(self):
#         # update internals
#         self.maxLong = self.pdfMaster['longitude'].max()
#         self.minLong = self.pdfMaster['longitude'].min()
#         self.maxLat = self.pdfMaster['latitude'].max()
#         self.minLat = self.pdfMaster['latitude'].min()
#         self.minAggVal = self.pdfMaster['avg(Values)'].min()
#         self.maxAggVal = self.pdfMaster['avg(Values)'].max()
        
#         # update center of map
#         self.center = [(self.minLat + self.maxLat)/2, (self.minLong + self.maxLong)/2]
#         self.zoom = 6
#         self.m = Map(default_tiles=TileLayer(opacity=1.0), center=self.center, zoom=self.zoom)
        
#         # loop over all the points in given dataframe, adding them to self.map
#         circles = []
#         for index,row in self.pdfMaster.iterrows():
#             _lat=row['latitude']
#             _long=row['longitude']
#             _count=row['count(Measurement)']
#             _coef=row['avg(Values)']
# #             pdb.set_trace()
#             # taking sqrt of count so that the  area of the circle corresponds to the count
#             c = Circle(location=(_lat,_long), radius=int(1200*np.sqrt(_count+0.0)), weight=1,
#                     color='#AAA', opacity=0.8, fill_opacity=0.4,
#                     fill_color=self.get_color(_coef))
#             circles.append(c)
#             self.m.add_layer(c)
#         self.m
        
    
#     def color_legend(self):
#         self.cfig = figure(figsize=[10,1])
#         ax = plt.subplot(111)
#         vals = self.cmap(np.arange(1,0,-.005))[:,:3]
#         vals3 = np.stack([vals]*10)
#         vals3.shape
#         ax.imshow(vals3)
#         midpoint = 200. * -self.minAggVal/(self.maxAggVal - self.minAggVal)
#         xticks((0,midpoint,200),["%4.1f"%v for v in (self.minAggVal,0.,self.maxAggVal)])
#         yticks(());

#     def get_color(self, val):
#         x = (val - self.minAggVal)/(self.maxAggVal - self.minAggVal)
#         return(rgb2hex(self.cmap(x)[:3]))<jupyter_output><empty_output>
{'fraction_non_alphanumeric': 0.11576470588235294, 'fraction_numerical': 0.008470588235294117, 'mean_word_length': 3.289866774323779, 'pattern_counts': {'":': 0, '<': 31, '<?xml version=': 0, '>': 31, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 1, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'chain_length': '7', 'dir': 'jupyter-structured-clean-dedup', 'id': '209070', 'license': 'MIT', 'n_tokens_mistral': 3470, 'n_tokens_neox': 3303, 'n_words': 947, 'path': 'HW3/PlainsWeatherStationLocations.ipynb', 'repo_name': 'timtyree/dsc291team4', 'size': '10625'}