变换RDD到有效的输入,使用包含与下面的代码CSV文件的目录的火花mllib算法k均值
我计算TF和IDF:变换RDD到有效的输入,使用包含与下面的代码CSV文件的目录的火花mllib算法k均值
import argparse
from os import system
### args parsing
parser = argparse.ArgumentParser(description='runs TF/IDF on a directory of
text docs')
parser.add_argument("-i","--input", help="the input in HDFS",
required=True)
parser.add_argument("-o", '--output', help="the output in HDFS",
required=True)
parser.add_argument("-mdf", '--min_document_frequency', default=1)
args = parser.parse_args()
docs_dir = args.input
d_out = "hdfs://master:54310/" + args.output
min_df = int(args.min_document_frequency)
# import spark-realated stuff
from pyspark import SparkContext
from pyspark.mllib.feature import HashingTF
from pyspark.mllib.feature import IDF
sc = SparkContext(appName="TF-IDF")
# Load documents (one per line).
documents = sc.textFile(docs_dir).map(lambda title_text:
title_text[1].split(" "))
hashingTF = HashingTF()
tf = hashingTF.transform(documents)
# IDF
idf = IDF().fit(tf)
tfidf = idf.transform(tf)
#print(tfidf.collect())
#save
tfidf.saveAsTextFile(d_out)
使用
print(tfidf.collect())
我得到这个输出:
[SparseVector(1048576, {812399: 4.3307}), SparseVector(1048576, {411697:
0.0066}), SparseVector(1048576, {411697: 0.0066}), SparseVector(1048576,
{411697: 0.0066}), SparseVector(1048576, {411697: 0.0066}), ....
我也有te STED的k均值mllib算法:
from __future__ import print_function
import sys
import numpy as np
from pyspark import SparkContext
from pyspark.mllib.clustering import KMeans
runs=4
def parseVector(line):
return np.array([float(x) for x in line.split(' ')])
if __name__ == "__main__":
if len(sys.argv) != 3:
print("Usage: kmeans <file> <k>", file=sys.stderr)
exit(-1)
sc = SparkContext(appName="KMeans")
lines = sc.textFile(sys.argv[1])
data = lines.map(parseVector)
k = int(sys.argv[2])
model = KMeans.train(data, k, runs)
print("Final centers: " + str(model.clusterCenters))
print("Total Cost: " + str(model.computeCost(data)))
sc.stop()
这个样本测试案例
0.0 0.0 0.0
0.1 0.1 0.1
0.2 0.2 0.2
9.0 9.0 9.0
9.1 9.1 9.1
9.2 9.2 9.2
,它工作正常。
现在我想在KMeans算法中应用上述tfidf的rdd输出,但我不知道如何像上面的示例文本那样转换rdd,或者如何正确拆分KMeans算法中的rdd正常工作。
我真的需要这个帮助。
UPDATE
我真正的问题是我如何可以读取输入到它从一个文本文件应用到KMEANS mllib这样
(1048576,[155412,857472,756332],[1.75642010278,2.41857747478,1.97365255252])
(1048576,[159196,323305,501636],[2.98856378408,1.63863706713,2.44956728334])
(1048576,[135312,847543,743411],[1.42412015238,1.58759872958,2.])
UPDATE2
我不是肯定在所有,但我认为我需要从矢量上面到下面的阵列,以便直接将它应用到KMeans mllib算法
1.75642010278 2.41857747478 1.97365255252
2.98856378408 1.63863706713 2.44956728334
1.42412015238 1.58759872958 2.
IDF的输出是一个数据帧SparseVector
。 KMeans将向量作为输入(稀疏或密集),因此,不应该进行任何转换。您应该能够直接使用IDF的输出列作为KMeans的输入。
如果您需要在运行TFIDF和KMeans之间将数据保存到磁盘,我建议通过dataframe API将其保存为csv。
首先转换成数据帧使用:
from pyspark.sql import Row
row = Row("features") # column name
df = tfidf.map(row).toDF()
的另一种方式,而不输入转换:
df = tfidf.map(lambda x: (x,)).toDF(["features"])
转换后的数据帧保存为拼花文件:
df.write.parquet('/path/to/save/file')
要读取数据,只需使用:
from pyspark.sql import SQLContext
sqlContext = SQLContext(sc)
df = sqlContext.read.parquet('/path/to/file')
# converting from dataframe into an RDD[Vector]
data = df.rdd.map(list)
如果您在任何情况下,需要从保存为一个字符串向量转换,这也是可能的。以下是一些示例代码:
from pyspark.mllib.linalg import Vectors, VectorUDT
from pyspark.sql.functions import udf
df = sc.parallelize(["(7,[1,2,4],[1,1,1])"]).toDF(["features"])
parse = udf(lambda s: Vectors.parse(s), VectorUDT())
df.select(parse("features"))
首先使用相同的格式创建示例数据框。然后使用UDF
将字符串解析为向量。如果你想要一个rdd而不是数据框,可以使用上面的代码在“从parquet读取”部分进行转换。
但是,IDF
的输出非常稀疏。矢量的长度为1048576,其中只有一个值大于1.KMeans不会给你任何有趣的结果。
我建议你看看word2vec。它会给你一个更紧凑的矢量为每个单词和聚类这些载体会更有意义。使用这种方法,您可以接收一个单词映射到它们可以用于聚类的矢量表示。
Thx为你的回复,但是当我使用tfidf的输出(顺便保存为文本文件,你可以在tf idf代码的最后一行看到)我得到这个errorTypeError:无法转换类型到Vector中。 –
Gmilios
@Gmilios我看到了,所以基本上你的问题是保存来自tfidf的数据,然后在KMeans之前再次读取它? – Shaido
@Gmilios增加了一些关于这个答案。 – Shaido