我有一个CSV文件,格式如下:
product_id1,product_title1 product_id2,product_title2 product_id3,product_title3 product_id4,product_title4 product_id5,product_title5 [...]
product_idX是一个整数,product_titleX是一个String,例如:
453478692, Apple iPhone 4 8Go
我正在尝试从我的文件创建TF-IDF,所以我可以将它用于MLlib中的朴素贝叶斯分类器.
到目前为止,我正在使用Spark for Scala并使用我在官方页面和Berkley AmpCamp 3和4上找到的教程.
所以我正在读文件:
val file = sc.textFile("offers.csv")
然后我将它映射到元组中 RDD[Array[String]]
val tuples = file.map(line => line.split(",")).cache
在我将元组转换成对之后 RDD[(Int, String)]
val pairs = tuples.(line => (line(0),line(1)))
但我被困在这里,我不知道如何从它创建Vector,把它变成TFIDF.
谢谢
为了自己这样做(使用pyspark),我首先从语料库中创建两个数据结构.第一个是关键的价值结构
document_id, [token_ids]
第二个是反向索引
token_id, [document_ids]
我将分别称为语料库和inv_index.
为了得到这个,我们需要计算每个文档中每个标记的出现次数.所以
from collections import Counter
def wc_per_row(row):
cnt = Counter()
for word in row:
cnt[word] += 1
return cnt.items()
tf = corpus.map(lambda (x, y): (x, wc_per_row(y)))
df只是每个术语倒排索引的长度.由此我们可以计算出idf.
df = inv_index.map(lambda (x, y): (x, len(y)))
num_documnents = tf.count()
# At this step you can also apply some filters to make sure to keep
# only terms within a 'good' range of df.
import math.log10
idf = df.map(lambda (k, v): (k, 1. + log10(num_documents/v))).collect()
现在我们只需要在term_id上进行连接:
def calc_tfidf(tf_tuples, idf_tuples):
return [(k1, v1 * v2) for (k1, v1) in tf_tuples for
(k2, v2) in idf_tuples if k1 == k2]
tfidf = tf.map(lambda (k, v): (k, calc_tfidf(v, idf)))
不过,这不是一个特别高效的解决方案.调用collect将idf带入驱动程序,以便它可用于连接似乎是错误的事情.
当然,它需要首先标记并创建从词汇表中的每个uniq标记到某个token_id的映射.
如果有人能改进这一点,我很感兴趣.