<SYSTEM_TASK:>
Given the following text description, write Python code to implement the functionality described below step by step
<END_TASK>
<USER_TASK:>
Description:
 Example 5
Step1:  Each of these lines contains 6 semi-colon delimited columns
Step2:  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
Step3:  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 time series analysis isn't distorted.  
PySpark's ".distinct()" method accomplishes this; we issue a ".collect()" as well to assign the RDD's values to a variable.
Step4:  Now we create an ordered dictionary to preserve the metric (and consequently, column) ordering.  If we did not create this OrderedDict, the keys' ordering may be permuted.  This will render ML techniques useless.
The object is broadcast to all executors to be used in a future mapped function.
Step5:  The two functions below are adapted from user patricksurry's answer to this Stack Overflow question
Step6:  We need to perform an aggregation by key.  This operation takes two functions as arguments
Step7:  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.
Step8:  Let's take a look at the results.
Step9:  To sort the entire RDD, we use a sortByKey.
<ASSISTANT_TASK:>
Python Code:
from pyspark import SparkConf, SparkContext
from collections import OrderedDict
partitions = 48
parcsv = sc.textFile("/lustre/janus_scratch/dami9546/lustre_timeseries.csv", partitions)
parcsv.take(5)
Explanation: Example 5: 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 some machine learning methods.
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)
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 time series analysis 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
ordered = sc.broadcast(OrderedDict(sorted(basedict.items(), key=lambda y: y[0])))
Explanation: Now we create an ordered dictionary to preserve the metric (and consequently, column) ordering.  If we did not create this OrderedDict, the keys' ordering may be permuted.  This will render ML techniques useless.
The object is broadcast to all executors to be used in a future mapped function.
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):
  tmp = ordered.value.copy()
  tmp.update(new[1])
  return new[0], tmp
pivoted = aggregated.map(mergedicts)
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
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
<END_TASK>