from __future__ import print_function

# from pyspark import SparkContext, SparkConf
# from pyspark.mllib.clustering import KMeans, KMeansModel

# # http://spark.apache.org/docs/2.0.0/api/python/pyspark.mllib.html#pyspark.mllib.evaluation.RankingMetrics
# from pyspark.mllib.evaluation import BinaryClassificationMetrics, MulticlassMetrics, RankingMetrics
import numpy as np
from numpy import array
import pandas as pd

from random import randrange
from math import sqrt

!ls -l

sc

conf = SparkConf()

# sc = SparkContext("local", "Simple App")

# sc.stop(sc)
# sc.getOrCreate("local", "Simple App")

"""
Explanation: K-means in PySpark
The next machine learning method I'd like to introduce is about clustering, K-means. It is an unsupervised learning method where we would like to group the observations into K groups (or subsets). We call it "unsupervised" since we don't have associated response measurements together with the observations to help check and evaluate the model we built (of course we can use other measures to evaluate the clustering models).
K-means may be the simplest approach for clustering while it’s also an elegant and efficient method. To produce the clusters, K-means method only requires the number of clusters K as its input.
The idea of K-means clustering is that a good clustering is with the smallest within-cluster variation (a measurement of how different the observations within a cluster are from each other) in a possible range. To achieve this purpose, K-means algorithm is designed in a "greedy" algorithm fashion
K-means Algorithm
1. For each observation, assign a random number which is generated from 1 to *K* to it.

2. For each of the *K* clusters, compute the cluster center. The *k*th cluster’s center is the vector of the means of the vectors of all the observations belonging to the kth cluster.

3. Re-assign each observation to the cluster whose cluster center is closest to this observation.

4. Check if the new assignments are the same as the last iteration. If not, go to step 2; if yes, END.

An example of iteration with K-means algorithm is presented below

Now it's time to implement K-means with PySpark. I generate a dateset myself, it contains 30 observations, and I purposedly "made" them group 3 sets.
troubleshooting:
- More: https://www.dataquest.io/blog/pyspark-installation-guide/
- https://stackoverflow.com/questions/23280629/multiple-sparkcontexts-error-in-tutorial
- check this tutorial he based it off of: http://blog.insightdatalabs.com/jupyter-on-apache-spark-step-by-step/
Dependencies


download for py27:
bash
conda install numpy
conda install pandas


install numpy on the nodes:

in jupyter: https://blog.cloudera.com/blog/2015/09/how-to-prepare-your-apache-hadoop-cluster-for-pyspark-jobs/
https://docs.continuum.io/anaconda-cluster/manage-cluster
End of explanation
"""


transactions = pd.read_csv('transactions.csv')
transactions['Date'] = pd.to_datetime(transactions['Date'],unit='ms') #coerce date format
transactions[:3]

print('transactions columns: ', list(transactions.columns))

"""
Explanation: Get Numpy on the cluster

required for Kmeans

from local machine:
```bash
install dependencies on the cluster ==================================
flintrock run-command test-cluster 'sudo yum install -y gcc'
have faith...flintrock takes awhile (3-5 min)
flintrock run-command test-cluster 'pip install --user numpy' 
```
note to self: learn how to run bash scripts
Real Transaction Data
GraphX in PySpark -----------------------------------------------------

https://stackoverflow.com/questions/23302270/how-do-i-run-graphx-with-python-pyspark
GraphFrames is the way to go
Work flow: http://aosc.umd.edu/~ide/data/teaching/amsc663/14fall/amsc663_14proposalpresentation_stefan_poikonen.pdf
Pandas DataFrames & features --> graphframes
graph analytics --> export as features



Real Transaction Data
End of explanation
"""


#establish sql context
from pyspark.sql import SQLContext

# Instantiate SQL_SparkContext object
SQL_CONTEXT = SQLContext(sc)

nodes = pd.read_csv('transactions.csv',
                    usecols=['Source', 'isTainted'],
                    low_memory=True,
                    iterator=True,
                    chunksize=1000)

# Concatenate chunks into list & convert to DataFrame
nodes = pd.DataFrame(pd.concat(list(nodes), ignore_index=True))

# Create a Vertex DataFrame with unique ID column "id"
nodes.columns = ['id', 'isTainted']
print('nodes columns: ', list(nodes.columns))

NODES = SQL_CONTEXT.createDataFrame(nodes)
NODES.take(3)

# Parallelize -----------------------------------------------------
# VERTICES = sc.parallelize(nodes)

# data = array([observation_group_1, observation_group_2, observation_group_3]).reshape(n_in_each_group*3, 5)
# data

# data = sc.parallelize(data)

# data.getNumPartitions()

# # Generate the observations -----------------------------------------------------
# n_in_each_group = 10   # how many observations in each group
# n_of_feature = 5 # how many features we have for each observation

# observation_group_1=[]
# for i in range(n_in_each_group*n_of_feature):
# 	observation_group_1.append(randrange(5, 8))

# observation_group_2=[]
# for i in range(n_in_each_group*n_of_feature):
# 	observation_group_2.append(randrange(55, 58))

# observation_group_3=[]
# for i in range(n_in_each_group*n_of_feature):
# 	observation_group_3.append(randrange(105, 108))

# del GENES_DF_CLEAN, GENES_DF, GENES

"""
Explanation: DataFrame --> GraphFrame
from: http://www.datareply.co.uk/blog/2016/9/20/running-graph-analytics-with-spark-graphframes-a-simple-example
Loading the Data - Nodes
We will use the pandas package for loading the dataset as a dataframe.
- Though not explicitly necessary in this case, the code I am using below accounts for cases where the file size is too big to be loaded in one go and performs the loading operation in chunks.
- We first load the genes with the following code.
- Notice that since the .csv file contains relationships, genes under the OFFICIAL_SYMBOL_A header may appear multiple times and thus duplicates need to be removed for unique identifiers
- Make sure to name the dataframe's single column that we are importing with the header "id" otherwise you will get an exception when you try to instantiate the GraphFrame based on it.
End of explanation
"""


edges = pd.read_csv('transactions.csv',
                    usecols=['Amount $', 'Date', 'Destination', 'Source', 'Transaction ID', 'isTainted'],
                    low_memory=True,
                    iterator=True,
                    chunksize=1000)

# Concatenate chunks into list & convert to DataFrame
edges = pd.DataFrame(pd.concat(list(edges), ignore_index=True))

cols = ['Source', 'Destination', 'isTainted', 'Amount $', 'Date', 'Transaction ID']

edges = edges[cols]
print('edges columns: ', list(edges.columns))

# Create an Edge DataFrame with "src" and "dst" columns
edges.columns = ["src", "dst", "relationship", 'Amount $', 'Date', 'TxID']
print('edges columns: ', list(edges.columns))

EDGES = SQL_CONTEXT.createDataFrame(edges)
EDGES.take(3)

"""
Explanation: Loading the Data - Edges
We repeat the same process as before with the difference that now we will be importing all three columns. If I am to be strict, I should say that this step is not really necessary if one opts for loading everything in one go the first time, then SELECTing different column-dataframes thus efficiently differentiating between vertices and edges. Still, I am repeating the code here for presentation convenience. Also, even though it is not really necessary in our case, you should make sure that once you are done with the loading process you get rid of anything that should not be kept in the memory any more, e.g. run:
spark
del GENES_DF_CLEAN, GENES_DF, GENES
Side note: The del statement does not directly reclaim memory. It only, removes a reference, which decrements the reference count on the value and if that count is zero, the memory can be reclaimed. In general, CPython will reclaim the memory immediately, there's no need to wait for the garbage collector to run.
Here is the code for loading the edges:
End of explanation
"""


from graphframes import *

type(NODES), type(EDGES)

# Next we finally create the graph:
g = GraphFrame(NODES, EDGES)

"""
Explanation: Make the graph
End of explanation
"""


# Query: Count the number of "isTainted" connections in the graph.
print(g.vertices.count())
print(g.edges.count())
print(g.degrees.count())
print(g.vertices.filter("isTainted = 5").count())

"""
Explanation: Graph Analytics

More examples: http://graphframes.github.io/user-guide.html
MasterNodePublicDNS: http://ec2-54-213-254-250.us-west-2.compute.amazonaws.com:8080/

Descriptive Statistics
End of explanation
"""


# Query: Get in-degree of each vertex.
print("Vertex in-Degree -----------------------------------------------------------------------")
df = g.inDegrees.sort('inDegree', ascending=False).toPandas()
transactions = transactions.merge(df,
                                  left_on='Source',
                                  right_on='id',)
transactions.head()

print("Vertex out-Degree ----------------------------------------------------------------------")
df = g.outDegrees.sort('outDegree', ascending=False).toPandas()
transactions = transactions.merge(df,
                                  left_on='Source',
                                  right_on='id')
transactions.head()

print("Vertex degree --------------------------------------------------------------------------")
df = g.degrees.sort('degree', ascending=False).toPandas()
transactions = transactions.merge(df,
                                  left_on='Source',
                                  right_on='id')
transactions.head()

transactions = transactions.drop(['id_x', 'id_y', 'id'], axis = 1)

"""
Explanation: Put a feature in pandas - put it back into pandas if resorting to sklearn
End of explanation
"""


# hits no space left on device
print("Triangle Count -------------------------------------------------------------------------")
RESULTS = g.triangleCount()
df = RESULTS.select("id", "count").toPandas()
transactions = transactions.merge(df,
                                  left_on='Source',
                                  right_on='id')
transactions.head()

print("Label Propagation ----------------------------------------------------------------------")
# Convergence is not guaranteed
df = g.labelPropagation(maxIter=10).toPandas()
# transactions = transactions.merge(df,
#                                   left_on='Source',
#                                   right_on='id')

df.head()

"""
Explanation: Extra Queries - THIS IS WHERE THINGS BREAK
End of explanation
"""


# # Run PageRank algorithm the other one, and show results.
# results = g.pageRank(resetProbability=0.01, maxIter=20)
# results.vertices.select("id", "pagerank").show()

# Run PageRank algorithm (takes awhile), and show results.
print("PageRank -------------------------------------------------------------------------------")
df = g.pageRank(resetProbability=0.15, tol=0.01)\
  .vertices.sort('pagerank', ascending=False).toPandas()
# transactions = transactions.merge(df,
#                                   left_on='Source',
#                                   right_on='id')

df.head()

print("Find Shortest Paths w.r.t. Tainted Wallets ---------------------------------------------------")
SHORTEST_PATH = g.shortestPaths(landmarks=["5"])
df = SHORTEST_PATH.select("id", "distances").toPandas()
# transactions = transactions.merge(df,
#                                   left_on='Source',
#                                   right_on='id')
df.head()

"""
Explanation: There are two implementations of PageRank.

The first implementation uses the standalone GraphFrame interface and runs PageRank for a fixed number of iterations. This can be run by setting maxIter.
The second implementation uses the org.apache.spark.graphx.Pregel interface and runs PageRank until convergence. This can be run by setting tol.
Both implementations support non-personalized and personalized PageRank, where setting a sourceId personalizes the results for that vertex.
End of explanation
"""


print('transactions columns: ', list(transactions.columns))

cols = ['inDegree', 'outDegree', 'degree']
tmp_transactions = transactions[cols]

def string_to_int(value):
    try:
        return int(value)
    except ValueError:  
        return None

for column in tmp_transactions.columns:
    tmp_transactions[column] = tmp_transactions[column].apply(string_to_int)

tmp_transactions.info()

data = sc.parallelize(tmp_transactions)

data.getNumPartitions()

# for MultinomialNB classification
tmp_transactions.to_csv('tmp-txns-no-headers.txt', header=False, index = True)

!ls

"""
Explanation: Porting the graphframe (if necessary)

http://graphframes.github.io/user-guide.html#saving-and-loading-graphframes

Cluster like your life depends on it
End of explanation
"""


# Build the K-Means model
# the initializationMode can also be "k-means||" or set by users.
clusters = KMeans.train(data, 2, maxIterations=3, initializationMode="random")

# Collect the clustering result
result=data.map(lambda point: clusters.predict(point)).collect()
print(result)

# 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.map(lambda point: error(point)).reduce(lambda x, y: x + y)
print("Within Set Sum of Squared Error = " + str(WSSSE))

"""
Explanation: ## Run the K-Means algorithm -----------------------------------------------------
End of explanation
"""


from pyspark.mllib.clustering import PowerIterationClustering, PowerIterationClusteringModel

# Load and parse the data
data = sc.textFile("tmp-txns-no-headers.txt")
similarities = data.map(lambda line: tuple([float(x) for x in line.split(',')]))

type(similarities)

# Cluster the data into two classes using PowerIterationClustering
model = PowerIterationClustering.train(similarities, k=2, maxIterations=10, initMode='random')

results = []
assignments = model.assignments().collect()
assignments[:3]

for x in assignments:
    results.append([x.id, x.cluster])

results[:3]

# Save and load model
# model.save(sc, "PICModel")
# sameModel = PowerIterationClusteringModel.load(sc, "PICModel")

results_df = pd.DataFrame(results, index=None, columns = ['id', 'cluster'])

# DOUBLE BRACKETS CREATES A DF
merged = transactions[['isTainted']].merge(results_df, left_index=True, right_on='id')

"""
Explanation: PowerIteration Clustering -----------------------------------------------------
Power Iteration Clustering
- https://spark.apache.org/docs/2.1.0/mllib-clustering.html#power-iteration-clustering-pic
- http://www.cs.cmu.edu/~wcohen/postscript/icml2010-pic-final.pdf
columns are : id,inDegree,outDegree,degree
End of explanation
"""


import pickle
with open("cluster-results.pkl", 'rb') as picklefile: 
    results = pickle.load(picklefile)

results.head(3)

def convert_to_true(value):
    if value == 5:
        return 1
    else:
        return 0

results['isTainted'] = results['isTainted'].apply(convert_to_true)

results[results['isTainted'] == 1].head(3)

from sklearn.metrics import classification_report, homogeneity_score, completeness_score

y_true = results['isTainted']
y_pred = results['cluster']
target_names = ['TaintedWallet', 'Wallet']
print(classification_report(y_true, y_pred, target_names=target_names))

"""
Explanation: Model Evaluation on Local Machine

Evaluating the performance of a clustering algorithm is not as trivial as counting the number of errors or the precision and recall of a supervised classification algorithm.
In particular any evaluation metric should not take the absolute values of the cluster labels into account, but rather, if this clustering defines separations of the data similar to some ground truth set of classes
OR satisfying some assumption such that (members belonging to the same class) are more similar that (members of different classes) according to some similarity metric.
End of explanation
"""


# homogeneity: each cluster contains only members of a single class.
homogeneity_score(y_true, y_pred)

# completeness: all members of a given class are assigned to the same cluster.
completeness_score(y_true, y_pred)

"""
Explanation: Conditional entropy analyses on clusters
End of explanation
"""


# http://spark.apache.org/docs/2.0.0/api/python/pyspark.mllib.html#pyspark.mllib.classification.NaiveBayesModel
from pyspark.mllib.classification import NaiveBayes, NaiveBayesModel
from pyspark.mllib.regression import LabeledPoint
from pyspark.mllib.linalg import SparseVector


data = [LabeledPoint(0.0, [0.0, 0.0]),
        LabeledPoint(0.0, [0.0, 1.0]),
        LabeledPoint(1.0, [1.0, 0.0])]

model = NaiveBayes.train(sc.parallelize(data))
model.predict(array([0.0, 1.0]))
model.predict(array([1.0, 0.0]))

model.predict(sc.parallelize([[1.0, 0.0]])).collect()

sparse_data = [LabeledPoint(0.0, SparseVector(2, {1: 0.0})),
               LabeledPoint(0.0, SparseVector(2, {1: 1.0})),
               LabeledPoint(1.0, SparseVector(2, {0: 1.0}))]

model = NaiveBayes.train(sc.parallelize(sparse_data))
model.predict(SparseVector(2, {1: 1.0}))

model.predict(SparseVector(2, {0: 1.0}))

"""
Explanation: supervised learning techniques are rigid and don't take into account changing patterns.
Using clustering techniques in an unsupervised manner allows for adaptive fraud detection even if fraudsters change the behaviors that led to classification in the first place.

MultinomialNB Classification - tip from guo

possibly to determine feature importances?
End of explanation
"""


import os, tempfile
path = tempfile.mkdtemp()

model.save(sc, path)
sameModel = NaiveBayesModel.load(sc, path)
sameModel.predict(SparseVector(2, {0: 1.0})) == model.predict(SparseVector(2, {0: 1.0}))

from shutil import rmtree
try:
    rmtree(path)
    except OSError:
        pass

"""
Explanation: DOESNT WORK: Exporting the model
End of explanation
"""


from graphframes.examples import Graphs
g = Graphs(sqlContext).friends()  # Get example graph

# Search for pairs of vertices with edges in both directions between them.
motifs = g.find("(a)-[e]->(b); (b)-[e2]->(a)")
motifs.show()

# More complex queries can be expressed by applying filters.
motifs.filter("b.age > 30").show()

"""
Explanation: FINISH LATER
Pairwise similarity matrix for PIC:

http://www.cs.cmu.edu/~wcohen/postscript/icml2010-pic-final.pdf
http://scikit-learn.org/stable/auto_examples/cluster/plot_mini_batch_kmeans.html
http://scikit-learn.org/stable/modules/generated/sklearn.metrics.pairwise.cosine_similarity.html

if clustering on csvs, cut the frame down and use sklearn...

Other options:
https://github.com/SsureyMoon/Bitcoin-network-cluster 
https://github.com/JeremyRubin/BTCSpark
https://github.com/mikispag/bitiodine

for another time: Spark-YARN & performance tuning (set AMIs to m3.mediums)

https://docs.continuum.io/anaconda-scale/howto/spark-yarn
https://blog.cloudera.com/blog/2015/03/how-to-tune-your-apache-spark-jobs-part-2/

Motif finding

http://graphframes.github.io/user-guide.html#motif-finding
End of explanation
"""


