%matplotlib inline
import pandas as pd
import seaborn as sns
import numpy as np

data = pd.read_csv("https://raw.githubusercontent.com/pydata/pandas/master/pandas/tests/data/iris.csv")
data.head()

"""
Explanation: KMeans: Scitkit, Pilot and Spark/MLlib
This is perhaps the best known database to be found in the pattern recognition literature. The data set contains 3 classes of 50 instances each, where each class refers to a type of iris plant (see https://archive.ics.uci.edu/ml/datasets/Iris). 
Source: R. A. Fisher, The Use of Multiple Measurements in Taxonomic Problems, 1936, http://rcs.chemometrics.ru/Tutorials/classification/Fisher.pdf
Pictures (Source Wikipedia)
<table>
<tr><td>
Setosa
</td><td>
Versicolor
</td><td>
Virginica
</td></tr>
<tr><td>
<img src="https://upload.wikimedia.org/wikipedia/commons/thumb/5/56/Kosaciec_szczecinkowaty_Iris_setosa.jpg/180px-Kosaciec_szczecinkowaty_Iris_setosa.jpg"/> 
</td><td>
<img src="https://upload.wikimedia.org/wikipedia/commons/thumb/4/41/Iris_versicolor_3.jpg/320px-Iris_versicolor_3.jpg"/>
</td><td>
<img src="https://upload.wikimedia.org/wikipedia/commons/thumb/9/9f/Iris_virginica.jpg/295px-Iris_virginica.jpg"/>
</td></tr></table>

1. Data Overview
We will begin by loading the data into a Pandas dataframe.
End of explanation
"""


from sklearn.cluster import KMeans
kmeans = KMeans(n_clusters=3)
results = kmeans.fit_predict(data[['SepalLength', 'SepalWidth', 'PetalLength', 'PetalWidth']])

data_kmeans=pd.concat([data, pd.Series(results, name="ClusterId")], axis=1)
data_kmeans.head()

"""
Explanation: 2. Scikit
Scikit is a machine learning library for Python built upon numpy and matplotlib. It provides functions for classification, regression, clustering and other common analytics tasks.
End of explanation
"""


print "Sum of squared error: %.1f"%kmeans.inertia_

sns.pairplot(data_kmeans, vars=["SepalLength", "SepalWidth", "PetalLength", "PetalWidth"], hue="ClusterId");

"""
Explanation: In the following we evaluate the resulting fit (commonly referred to as the model), using the sum of squared errors and a pair plot. The following pair plot shows the scatter-plot between each of the four features. Clusters for the different species are indicated by different colors.
End of explanation
"""


import os, sys
import commands
import radical.pilot as rp
os.environ["RADICAL_PILOT_DBURL"]="mongodb://ec2-54-221-194-147.compute-1.amazonaws.com:24242/giannis"

def print_details(detail_object):
    if type(detail_object)==str:
        detail_object = ast.literal_eval(detail_object)
    for i in detail_object:
        detail_object[i]=str(detail_object[i])
    #print str(detail_object)
    return pd.DataFrame(detail_object.values(), 
             index=detail_object.keys(), 
             columns=["Value"])


session = rp.Session()
c = rp.Context('ssh')
c.user_id = "radical"
session.add_context(c)
pmgr = rp.PilotManager(session=session)
umgr = rp.UnitManager (session=session,
                       scheduler=rp.SCHED_DIRECT_SUBMISSION)
print "Session id: %s Pilot Manager: %s" % (session.uid, str(pmgr.as_dict()))
pdesc = rp.ComputePilotDescription ()
pdesc.resource = "local.localhost_anaconda"
pdesc.runtime  = 10 
pdesc.cores    = 16
pdesc.cleanup  = False
pilot = pmgr.submit_pilots(pdesc)
umgr = rp.UnitManager (session=session,
                       scheduler=rp.SCHED_DIRECT_SUBMISSION)
umgr.add_pilots(pilot)

"""
Explanation: 3. Pilot Approach
We will now use RADICAL-Pilot to compute the distance function, as a simple representation of how the above example can be executed as a task-parallel application.
End of explanation
"""


number_clusters = 3
clusters = data.sample(number_clusters)

clusters

clusters.to_csv("clusters.csv")
data.to_csv("points.csv")

"""
Explanation: In the following, we will partition the data and distribute it to a set of CUs for fast processing
End of explanation
"""


def compute_new_centroids(distances):
    df = pd.DataFrame(distances)
    df[4] =  df[4].astype(int)
    df = df.groupby(4)[0,1,2,3].mean()
    centroids_np = df.as_matrix()
    return centroids_np

"""
Explanation: Helper Function for computing new centroids as mean of all points assigned to a cluster
End of explanation
"""


for i in range(10):
    distances =!/opt/anaconda/bin/python mapper.py points.csv clusters.csv
    distances_np =  np.array(eval(" ".join(distances)))
    new_centroids = compute_new_centroids(distances_np)
    new_centroids_df = pd.DataFrame(new_centroids, columns=["SepalLength", "SepalWidth", "PetalLength", "PetalWidth"])
    new_centroids_df.to_csv("clusters.csv")        

"""
Explanation: Running Mapper Function as an External Process
End of explanation
"""


import urlparse

def get_output(compute_unit):
    working_directory=compute_unit.as_dict()['working_directory']
    path = urlparse.urlparse(working_directory).path
    output=open(os.path.join(path, "STDOUT")).read()
    return output

"""
Explanation: Running Mapper Function inside RADICAL-Pilot
Helper function to read output from completed compute units after it has been executed inside the Pilot.
End of explanation
"""


for i in range(10):
    cudesc = rp.ComputeUnitDescription()
    cudesc.executable  = "/opt/anaconda/bin/python"
    cudesc.arguments   = [os.path.join(os.getcwd(), "mapper.py"),
                          os.path.join(os.getcwd(), "points.csv"),
                          os.path.join(os.getcwd(), "clusters.csv")]
    cu_set = umgr.submit_units([cudesc])
    umgr.wait_units()
    output = get_output(cu_set[0])
    distances_np =  np.array(eval(output))
    new_centroids = compute_new_centroids(distances_np)
    new_centroids_df = pd.DataFrame(new_centroids, columns=["SepalLength", "SepalWidth", "PetalLength", "PetalWidth"])
    new_centroids_df.to_csv("clusters.csv")        
    print "Finished iteration: %d"%(i)

"""
Explanation: This is the main application loop. The distance computation is executed inside a ComputeUnit.  See mapper.py for code. Data is read from files and written to stdout. We execute 10 iterations of KMeans.
End of explanation
"""


new_centroids_df

session.close()

"""
Explanation: Print out final centroids computed
End of explanation
"""


from numpy import array
from math import sqrt

%run ../env.py
%run ../util/init_spark.py

from pilot_hadoop import PilotComputeService as PilotSparkComputeService

try:
    sc
except:
    pilotcompute_description = {
        "service_url": "yarn-client://sc15.radical-cybertools.org",
        "number_of_processes": 5
    }
    pilot_spark = PilotSparkComputeService.create_pilot(pilotcompute_description=pilotcompute_description)

    sc = pilot_spark.get_spark_context()
    sqlCtx=SQLContext(sc)

"""
Explanation: Spark MLLib
In the following we utilize the Spark MLlib KMeans implementation. See http://spark.apache.org/docs/latest/mllib-clustering.html#k-means
We use Pilot-Spark to startup Spark.
End of explanation
"""


data_spark=sqlCtx.createDataFrame(data)
data_spark_without_class=data_spark.select('SepalLength', 'SepalWidth', 'PetalLength', 'PetalWidth').show()

"""
Explanation: Load and parse the data in a Spark DataFrame.
End of explanation
"""


data_spark_tuple = data_spark.map(lambda a: (a[0],a[1],a[2],a[3]))

from pyspark.mllib.clustering import KMeans, KMeansModel
clusters = KMeans.train(data_spark_tuple, 3, maxIterations=10,
                        runs=10, initializationMode="random")

# Evaluate clustering by computing Within Set Sum of Squared Errors
def error(point):
    center = clusters.centers[clusters.predict(point)]
    return sqrt(sum([x**2 for x in (point - center)]))

WSSSE = data_spark_tuple.map(lambda point: error(point)).reduce(lambda x, y: x + y)
print("Within Set Sum of Squared Error = " + str(WSSSE))

"""
Explanation: Convert DataFrame to Tuple for MLlib
End of explanation
"""


pilot_spark.cancel()

"""
Explanation: Stop Pilot-Job
End of explanation
"""


