变换RDD到有效的输入，使用包含与下面的代码CSV文件的目录的火花mllib算法k均值

Question

我计算TF和IDF：

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.

Answer 1

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。它会给你一个更紧凑的矢量为每个单词和聚类这些载体会更有意义。使用这种方法，您可以接收一个单词映射到它们可以用于聚类的矢量表示。