<gh_stars>1-10
import argparse

parser = argparse.ArgumentParser(description='runs kmeans on spark for .csv files')

parser.add_argument("-k","--K", help="the K parameter of K-means", type=int, required=True)
parser.add_argument("-mi","--max_iterations", help="the max iterations of the algorithm", type=int, required=True)
parser.add_argument("-i", "--input",  help="the input dir (RDD)", required=True)
parser.add_argument("-o", '--output', help="the output file", required=True )
parser.add_argument("-s", '--source', help="the input file source (eg scikit)", required=True )
args = parser.parse_args()



k = args.K
max_iter = args.max_iterations
fname = args.input
source = args.source
#if not fname.startswith('/'):
#    print "Please specify an absolute path for the input"
#    exit()

#fname = "hdfs://master:9000/"+fname
runs = 4 #???

from pyspark import SparkContext
from pyspark.mllib.clustering import KMeans
from numpy import array
from math import sqrt
from pyspark.mllib.linalg import SparseVector


sc = SparkContext( appName="kmeans")

def myVec(line):
    from pyspark.mllib.linalg import SparseVector
    return  eval("SparseVector"+line)


data = None 
# Load and parse the data
#try:
#    data = sc.textFile(fname).first()#.map(myVec)
#except Exception as e:
if source == "python": 
    import pickle, os
    print "Aw! python<3"
    os.system("hdfs dfs -copyToLocal %s /tmp/tfidfTMP"%fname)
    s = pickle.load(open("/tmp/tfidfTMP")).todense()
    #data = map(lambda k: k.tolist()[0], s)
    data = sc.parallelize(s, 100).map(lambda k: k.tolist()[0])
else:
    data = sc.textFile(fname).first()#.map(myVec)

# Build the model (cluster the data)
clusters = KMeans.train(data, k, maxIterations=max_iter, runs=runs, 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 = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y)
# print("Within Set Sum of Squared Error = " + str(WSSSE))

f = open(args.output, "w")
for c in clusters.clusterCenters:
    f.write("[")
    for i in range(len(c)):
        if c[i]>=0.1:
            f.write("(%d, %.3f)," % (i, c[i]))
    f.write("]\n")

f.close()

os.system("rm /tmp/tfidfTMP")

import argparse

parser = argparse.ArgumentParser(description='runs kmeans on spark for .csv files')

parser.add_argument("-k","--K", help="the K parameter of K-means", type=int, required=True)
parser.add_argument("-mi","--max_iterations", help="the max iterations of the algorithm", type=int, required=True)
parser.add_argument("-i", "--input",  help="the input dir (RDD)", required=True)
parser.add_argument("-o", '--output', help="the output file", required=True )
parser.add_argument("-s", '--source', help="the input file source (eg scikit)", required=True )
args = parser.parse_args()



k = args.K
max_iter = args.max_iterations
fname = args.input
source = args.source
#if not fname.startswith('/'):
#    print "Please specify an absolute path for the input"
#    exit()

#fname = "hdfs://master:9000/"+fname
runs = 4 #???

from pyspark import SparkContext
from pyspark.mllib.clustering import KMeans
from numpy import array
from math import sqrt
from pyspark.mllib.linalg import SparseVector


sc = SparkContext( appName="kmeans")

def myVec(line):
    from pyspark.mllib.linalg import SparseVector
    return  eval("SparseVector"+line)


data = None 
# Load and parse the data
#try:
#    data = sc.textFile(fname).first()#.map(myVec)
#except Exception as e:
if source == "python": 
    import pickle, os
    print "Aw! python<3"
    os.system("hdfs dfs -copyToLocal %s /tmp/tfidfTMP"%fname)
    s = pickle.load(open("/tmp/tfidfTMP")).todense()
    #data = map(lambda k: k.tolist()[0], s)
    data = sc.parallelize(s, 100).map(lambda k: k.tolist()[0])
else:
    data = sc.textFile(fname).first()#.map(myVec)

# Build the model (cluster the data)
clusters = KMeans.train(data, k, maxIterations=max_iter, runs=runs, 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 = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y)
# print("Within Set Sum of Squared Error = " + str(WSSSE))

f = open(args.output, "w")
for c in clusters.clusterCenters:
    f.write("[")
    for i in range(len(c)):
        if c[i]>=0.1:
            f.write("(%d, %.3f)," % (i, c[i]))
    f.write("]\n")

f.close()

os.system("rm /tmp/tfidfTMP")
#if not fname.startswith('/'): #    print "Please specify an absolute path for the input" #    exit() #fname = "hdfs://master:9000/"+fname #??? # Load and parse the data #try: #    data = sc.textFile(fname).first()#.map(myVec) #except Exception as e: #data = map(lambda k: k.tolist()[0], s) #.map(myVec) # Build the model (cluster the data) # # 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 = parsedData.map(lambda point: error(point)).reduce(lambda x, y: x + y) # print("Within Set Sum of Squared Error = " + str(WSSSE))