如何修复 pyspark 上的 TF.IDF 函数？

Question

我正在尝试通过 MapReduce（平台是 Databricks）在 pyspark 上开发 TF.IDF 进程。由于我对 pyspark、Databricks 和 MapReduce 的整个过程真的很陌生，所以我遇到了一些我无法解决的语法问题。

这是我尝试使用的代码运行:

full_text = sc.wholeTextFiles("/FileStore/tables/full_text.txt", 5) 
number_of_docs = full_text.count()

import re

def tokenize(s):
  return re.split("\W+", s.lower())

#Tokenize the text
tokenized_text = full_text.map(lambda text,title: (title, tokenize(text)))

#Count Words in each document
term_frequency = tokenized_text.flatMapValues(lambda x: x).countByValue()
term_frequency.items()[:20] # Display 20 lines

document_frequency = tokenized_text.flatMapValues(lambda x: x).distinct()\
                        .map(lambda title,word: (word,title)).countByKey()


document_frequency.items()[:10]

import numpy as np


def tf_idf(number_of_docs, term_frequency, document_frequency):
    result = []
    for key, value in tf.items():
        doc = key[0]
        term = key[1]
        df = document_frequency[term]
        if (df>0):
          tf_idf = float(value)*np.log(number_of_docs/df)

        result.append({"doc":doc, "score":tf_idf, "term":term})
    return result



tf_idf_output = tf_idf(number_of_docs, term_frequency, document_frequency)

tf_idf_output[:10]

问题在线：

term_frequency = tokenized_text.flatMapValues(lambda x: x).countByValue()

document_frequency = tokenized_text.flatMapValues(lambda x: x).distinct()\
                        .map(lambda title,word: (word,title)).countByKey()

错误：作业因阶段失败而中止：阶段 1.0 中的任务 0 失败 1 次，最近的失败：阶段 1.0 中丢失任务 0.0（TID 1、本地主机、执行程序驱动程序）

非常感谢你的帮助，因为我不知道如何解决这个问题:(

非常感谢

Answer 1

map 采用一元（单个参数）函数。您传递的是二进制（两个参数）函数。

回到 Python 2 天 lambdas 支持元组参数解包

~~lambda (text, title): (title, tokenize(text))~~

今天你应该索引

lambda tt: (tt[0], tokenize(tt[1]))

或更好地使用适当的功能：

def tuple_tokenizer(title_and_text):
    title, text = title_and_text
    return title, tokenize(text)

如何修复 pyspark 上的 TF.IDF 函数？

How to fix TF.IDF functions on pyspark?

python

mapreduce

apache-spark

pyspark