Come posso creare un TF-IDF per la classificazione del testo utilizzando Spark?

Question

Ho un file CSV con il seguente formato:

product_id1,product_title1 
product_id2,product_title2 
product_id3,product_title3 
product_id4,product_title4 
product_id5,product_title5 
[...] 

Il product_idX è un intero e l'product_titleX è una stringa, ad esempio:

453478692, Apple iPhone 4 8Go 

Sto cercando di creare il TF-IDF dal mio file quindi posso usarlo per un classificatore Naive Bayes in MLlib.

Sto utilizzando Spark per Scala finora e utilizzando i tutorial che ho trovato sulla pagina ufficiale e il Berkley AmpCamp 3 e 4.

Così sto leggendo il file:

val file = sc.textFile("offers.csv") 

Poi ho mappatura in tuple RDD[Array[String]]

val tuples = file.map(line => line.split(",")).cache 

e dopo sto trasformando le tuple in coppie RDD[(Int, String)]

val pairs = tuples.(line => (line(0),line(1))) 

Ma io sono bloccato qui e non so come creare il Vector fro m per trasformarlo in TFIDF.

Grazie

Answer 1

fare da solo (usando pyspark), ho iniziato con la creazione di strutture di due dati fuori del corpus. La prima è una struttura di chiusura, valore di

document_id, [token_ids] 

Il secondo è un indice invertito come

token_id, [document_ids] 

chiamerò rispettivamente quelle corpus e inv_index.

Per ottenere tf è necessario contare il numero di occorrenze di ciascun token in ciascun documento. So

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))) 

Il df è semplicemente la lunghezza dell'indice invertito di ciascun termine. Da ciò possiamo calcolare l'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() 

Ora non ci resta che fare un join sul 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))) 

Questa non è una soluzione particolarmente performante, però. Chiamare collect per portare idf nel programma driver in modo che sia disponibile per il join sembra la cosa sbagliata da fare.

E, naturalmente, richiede prima tokenizzazione e creazione di una mappatura da ciascun token uniq nel vocabolario ad alcuni token_id.

Se qualcuno può migliorare su questo, sono molto interessato.