### Launching PySpark
# pyspark --num-executors 2 --executor-memory 1500M --master yarn

##### state with the highest average temperature in summer - by Spark RDD ----------
### support functions
# data transformation: list --> (state, (teplota, 1))
def adjust_line(rlist):
    stat = rlist[9]
    tepl = (int(rlist[4])/10.0 - 32)*5/9
    return (stat, (tepl, 1))

# agregation of pairs (teplota, 1) --> (sum, count)
def sum_count(a, b):
    sumA = a[0]
    sumB = b[0]
    countA = a[1]
    countB = b[1]
    return (sumA + sumB, countA + countB)

# reading a file with temperatures
teploty_raw = sc.textFile('/user/pascepet/data/teplota')

# row --> list of values
teploty1 = teploty_raw.map(lambda r: r.split(',')) 

# keeping only rows with months 6-8 and existing tempreture (not empty)
teploty2 = teploty1.filter(lambda rlist: (rlist[1] in set('678')) & (rlist[4] != ''))

# extrating necessary data from a list of values
teploty3 = teploty2.map(adjust_line)

# agregation by states
teploty_staty = teploty3.reduceByKey(sum_count)

# average for every state: (state, (sum, count)) -> (state, average)
teploty_staty2 = teploty_staty.map(lambda x: (x[0], x[1][0]/x[1][1]))

# sorting by average temperature descending
teploty_staty3 = teploty_staty2.sortBy(lambda y: y[1], ascending=False)

# and printing...
teploty_staty3.take(1)
teploty_staty3.collect()
####################################
    

##### three ways how to read data as a DataFrame -------------------------------
### 1. directly from CSV
# requires external package when using older Spark version (1.6)
# pyspark --num-executors 2 --executor-memory 1500M --packages com.databricks:spark-csv_2.10:1.5.0 --master yarn
teploty_DF1 = sqlContext.read \
    .format("com.databricks.spark.csv") \
    .option("header", "true") \
    .option("delimiter", ",") \
    .option("inferSchema", "true") \
    .load("/user/pascepet/data/teplota")


### 2. reading form Hive (if the table exists)
teploty_DF2 = sqlContext.sql('select * from fel_bigdata.teplota')

### 3. converting an existing RDD
from pyspark.sql import Row
import re

def uprav_radek_df_row(r):
    rlist = r.split(',')
    return Row(stanice=rlist[0], mesic=int(rlist[1]), den=int(rlist[2]), hodina=int(rlist[3]), \
        teplota=None if rlist[4]=='' else float(rlist[4]), flag=rlist[5], latitude=float(rlist[6]), \
        longitude=float(rlist[7]), vyska=float(rlist[8]), stat=rlist[9], nazev=rlist[10])

# reading RDD from file, then converting
teploty_raw = sc.textFile('/user/pascepet/data/teplota')
teploty_prep = teploty_raw.filter(lambda line: not(re.match(r'stanice', line))) \
    .map(uprav_radek_df_row)
teploty_DF3 = sqlContext.createDataFrame(teploty_prep)


##### working with a DataFrame: state with the highest average temperature by Spark SQL -----
### 1. registration DataFrame as a temporary table
teploty_DF1.registerTempTable("tpDF")

# agregation by a SQL query
teploty_DF1_prum = sqlContext.sql("""select stat, avg((teplota/10.0-32)*5/9) as tepl_prum from tpDF
where mesic in (6,7,8)
group by stat order by tepl_prum desc""")
teploty_DF1_prum.show(10)


### 2. by SQL-like operations
# keeping non-empty data of summer months
teploty_DF_upr = teploty_DF1.filter((teploty_DF1['mesic']>=6) & (teploty_DF1['mesic']<=8)) \
    .select('stat','teplota').dropna()
# convert temperature Fahrenheit -> Celsius degrees
teploty_DF_upr = teploty_DF_upr.withColumn('teplota', (teploty_DF_upr['teplota']/10.0 - 32) * 5/9)
# agregation - averages by states
teploty_statyDF = teploty_DF_upr.groupBy('stat').avg('teplota') \
    .toDF('stat', 'tepl_prum')
# ordering by the average temperature 
teploty_statyDF = teploty_statyDF.orderBy(teploty_statyDF['tepl_prum'].desc())
# printing...
teploty_statyDF.show(10)

#############################