我有这个火花计划,我会尽量把它限制在相关的部分
# Split by delimiter ,
# If the file is in unicode, we need to convert each value to a float in order to be able to
# treat it as a number
points = sc.textFile(filename).map(lambda line: [float(x) for x in line.split(",")]).persist()
# start with K randomly selected points from the dataset
# A centroid cannot be an actual data point or else the distance measure between a point and
# that centroid will be zero. This leads to an undefined membership value into that centroid.
centroids = points.takeSample(False, K, 34)
#print centroids
# Initialize our new centroids
newCentroids = [[] for k in range(K)]
tempCentroids = []
for centroid in centroids:
tempCentroids.append([centroid[N] + 0.5])
#centroids = sc.broadcast(tempCentroids)
convergence = False
ncm = NCM()
while(not convergence):
memberships = points.map(lambda p : (p, getMemberships([p[N]], centroids.value, m)))
cmax = memberships.map(lambda (p, mus) : (p, getCMax(mus, centroids.value)))
# Memberships
T = cmax.map(lambda (p, c) : (p, getMemberships2([p[N]], centroids.value, m, delta, weight1, weight2, weight3, c)))
I = cmax.map(lambda (p, c) : (p, getIndeterminateMemberships([p[N]], centroids.value, m, delta, weight1, weight2, weight3, c)[0]))
F = cmax.map(lambda (p, c) : (p, getFalseMemberships([p[N]], centroids.value, m, delta, weight1, weight2, weight3, c)[0]))
# Components of new centroids
wTm = T.map(lambda (x, t) : ('onekey', scalarPow(m, scalarMult(weight1, t))))
#print "wTm = " + str(wTm.collect())
print "at first reduce"
sumwTm = wTm.reduceByKey(lambda p1, p2 : addPoints(p1, p2))
#print "sumwTm = " + str(sumwTm.collect())
wTmx = T.map(lambda (x, t) : pointMult([x[N]], scalarPow(m, scalarMult(weight1, t))))
print "adding to cnumerator list"
#print wTmx.collect()
cnumerator = wTmx.flatMap(lambda p: getListComponents(p)).reduceByKey(lambda p1, p2 : p1 + p2).values()
print "collected cnumerator, now printing"
#print "cnumerator = " + str(cnumerator.collect())
#print str(sumwTm.collect())
# Calculate the new centroids
sumwTmCollection = sumwTm.collect()[0][1]
cnumeratorCollection = cnumerator.collect()
#print "sumwTmCollection = " + str(sumwTmCollection)
#cnumeratorCollection =cnumerator.collectAsMap().get(0).items
print "cnumeratorCollection = " + str(cnumeratorCollection)
for i in range(len(newCentroids)):
newCentroids[i] = scalarMult(1 / sumwTmCollection[i], [cnumeratorCollection[i]])
centroids = newCentroids
# Test for convergence
convergence = ncm.test([centroids[N]], [newCentroids[N]], epsilon)
#convergence = True
# Replace our old centroids with the newly found centroids and repeat if convergence not met
# Clear out space for a new set of centroids
newCentroids = [[] for k in range(K)]
这个程序在我的本地机器上运行得很好,但在独立集群上运行时,它的行为并不像预期的那样。它不一定会抛出错误,但它所做的与我在本地机器上运行时收到的输出不同。集群和3个节点似乎运行良好。我觉得问题是我一直在更新
centroids
,这是一个python列表,每次通过
while-loop
。每个节点是否可能没有该列表的最新副本?我想是的,所以我试着用
broadcast variable
但这些不能更新(只读)。我还尝试使用
accumulator
但这些只是为了积累。我还尝试将python列表保存为hdfs上的一个文件,以便每个节点都可以访问,但这并不奏效。你认为我对这个问题的理解正确吗?这里还可能发生什么事吗?如何获得在本地机器上正常工作但在集群上无法正常工作的代码?
