from pyspark import SparkConf, SparkContext
from collections import OrderedDict
from functools import partial

partitions = 64
parcsv = sc.textFile("/user/milroy/lustre_timeseries.csv", partitions)
parcsv.take(5)

"""
Explanation: Example 2: A fast parallel pivot, or preparing for time series analysis
End of explanation
"""


filtered = parcsv.filter(lambda line: len(line.split(';')) == 6)

"""
Explanation: Each of these lines contains 6 semi-colon delimited columns: hostname, metric name, value reported, type, units, and Unix epoch time.  Can we assume all do?  The example data is an excerpt of one day of Lustre data, but we have hundreds of full days which may contain dropped writes and malformed data.  I'll apply a filter to the data to select all lines with six columns.
Sometimes it isn't evident whether filters are needed until a succeeding RDD action fails.
End of explanation
"""


def cast(line):
    try:
      val = float(str(line.split(';')[2]))
    except:
      val = 0.0
    return (int(line.split(';')[5]), line.split(';')[0], 
            line.split(';')[1], val)

parsed = filtered.map(cast)

"""
Explanation: As seen above, the lines are Unicode, but in anticipation of necessary transformations the timestamp and values will need to be cast to appropriate types.  We'll need to create a function that takes each line as an argument and returns a 4-tuple (quadruple?), organized to facilitate intuitive indexing.  Let's pick the following ordering: (timestamp, host, metric, value).  We don't need the other values, so they are discarded.
Since the values in the third column are currently Unicode, a try-except structure is used to attempt to cast them to floats.  If unsuccessful we set them to zero rather than NaN, since these don't work with the forthcoming eigendecomposition.
An alternative to the try-except would be to apply a filter for lines whose third column can't be cast as a float.  I haven't compared the performance between these two.
End of explanation
"""


columns = parsed.map(lambda x: x[2]).distinct().collect()
basedict = dict((metric, 0.0) for metric in columns)
ordered = OrderedDict(sorted(basedict.items(), key=lambda y: y[0]))

"""
Explanation: Metrics aren't reported continuously, nor are the monitoring systems flawless.  We need to assemble a unique set (dictionary) of metrics for the pivot, but they must be ordered to make sure the covariance structure (for PCA) isn't distorted.  
PySpark's ".distinct()" method accomplishes this; we issue a ".collect()" as well to assign the RDD's values to a variable.
End of explanation
"""


def combine(u1, u2):
  u1.update(u2)
  return u1

def sequential(u, v):
  if not u:
      u = {}
  u[v[2]] = v[3]
  return u

"""
Explanation: The two functions below are adapted from user patricksurry's answer to this Stack Overflow question: http://stackoverflow.com/questions/30260015/reshaping-pivoting-data-in-spark-rdd-and-or-spark-dataframes.  Beware, patricksurry's answer is predominantly serial!
End of explanation
"""


aggregated = parsed.keyBy(lambda row: (row[0], row[1])).aggregateByKey(
    None, sequential, combine)

"""
Explanation: We need to perform an aggregation by key.  This operation takes two functions as arguments: the sequential and combination functions.  The sequential op constructs a dictionary from (metric, value) in each row, and the combine op combines row dictionaries based on identical (timestamp, host) keys.
<img src="aggregateByKey.png">
End of explanation
"""


def mergedicts(new, canon):
  tmp = canon.copy()
  tmp.update(new[1])
  return new[0], tmp

"""
Explanation: Now we need to impose the structure of our OrderedDict on each aggregated key, value pair.  We create a new function to copy our canonical dictionary (of ordered keys, and 0.0 values) and update it with the dictionaries created in the aggregateByKey step.
End of explanation
"""


pivoted = aggregated.map(partial(mergedicts, canon=ordered))

"""
Explanation: Spark operations typically accept functions with exactly one argument.  Since we must pass our canonical OrderedDict to mergedicts, we use Python's partial to pass the dictionary as a keyword argument.
A more desirable way of making "ordered" available to each partition would be to create a SparkContext broadcast variable.  Unfortunately, broadcast variables currently do not support the .copy() method.
End of explanation
"""


final_ordered = pivoted.takeOrdered(10, key=lambda x: x[0])

final_ordered[0][0]

"""
Explanation: Let's take a look at the results.
End of explanation
"""


final_sorted = pivoted.sortByKey(keyfunc= lambda k: k[0])

final_dict = final_sorted.map(lambda row: row[1].values())

"""
Explanation: To sort the entire RDD, we use a sortByKey.
End of explanation
"""


final_dict.coalesce(2).saveAsTextFile("/home/milroy/pyspark/processed.txt")

"""
Explanation: Writing the lists to disk takes quite a long time.  This is not optimized for Hadoop, and not writing in parallel.  An exercise for the reader!
End of explanation
"""


spark-shell --master $MASTER --driver-memory 12g

import org.apache.spark.mllib.linalg.Matrix
import org.apache.spark.mllib.linalg.distributed.RowMatrix
import org.apache.spark.mllib.linalg.{Vector, Vectors}

val datafilePattern = "/user/milroy/pivoted.txt"
val lustreData = sc.textFile(datafilePattern).cache()

val vecData = lustreData.map(line => line.split(",").map(
        line => line.drop(1).dropRight(1)).map(
        v => v.toDouble)).map(arr => Vectors.dense(arr))
val rmat: RowMatrix = new RowMatrix(vecData)
val pc: Matrix = rmat.computePrincipalComponents(15)

"""
Explanation: Now on to Scala Spark for time series PCA
Now exit the pyspark shell, and run spark-shell with the following options.
End of explanation
"""


