我正在尝试在JSON文件上创建一个LDA模型。
使用JSON文件创建Spark上下文:
import org.apache.spark.sql.SparkSession
val sparkSession = SparkSession.builder
.master("local")
.appName("my-spark-app")
.config("spark.some.config.option", "config-value")
.getOrCreate()
val df = spark.read.json("dbfs:/mnt/JSON6/JSON/sampleDoc.txt")
显示df
应该显示DataFrame
display(df)
文本分词
import org.apache.spark.ml.feature.RegexTokenizer
// Set params for RegexTokenizer
val tokenizer = new RegexTokenizer()
.setPattern("[\\W_]+")
.setMinTokenLength(4) // Filter away tokens with length < 4
.setInputCol("text")
.setOutputCol("tokens")
// Tokenize document
val tokenized_df = tokenizer.transform(df)
这应该显示已经分词化的 tokenized_df
。
display(tokenized_df)
获取停用词列表
%sh wget http://ir.dcs.gla.ac.uk/resources/linguistic_utils/stop_words > -O /tmp/stopwords
可选:将停用词复制到tmp文件夹
%fs cp file:/tmp/stopwords dbfs:/tmp/stopwords
收集所有的 停用词
val stopwords = sc.textFile("/tmp/stopwords").collect()
过滤掉停用词。
import org.apache.spark.ml.feature.StopWordsRemover
// Set params for StopWordsRemover
val remover = new StopWordsRemover()
.setStopWords(stopwords) // This parameter is optional
.setInputCol("tokens")
.setOutputCol("filtered")
// Create new DF with Stopwords removed
val filtered_df = remover.transform(tokenized_df)
展示过滤后的df
应该可以验证stopwords
已被移除
display(filtered_df)
将单词出现的频率向量化
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.Row
import org.apache.spark.ml.feature.CountVectorizer
// Set params for CountVectorizer
val vectorizer = new CountVectorizer()
.setInputCol("filtered")
.setOutputCol("features")
.fit(filtered_df)
验证 vectorizer
vectorizer.transform(filtered_df)
.select("id", "text","features","filtered").show()
在这之后,我发现在将这个vectorizer
适配到LDA时出现了问题。我认为问题在于CountVectorizer
提供了稀疏向量,但是LDA需要密集向量。仍在努力解决这个问题。
以下是无法进行转换的异常情况:
import org.apache.spark.mllib.linalg.Vector
val ldaDF = countVectors.map {
case Row(id: String, countVector: Vector) => (id, countVector)
}
display(ldaDF)
异常:
org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 4083.0 failed 4 times, most recent failure: Lost task 0.3 in stage 4083.0 (TID 15331, 10.209.240.17): scala.MatchError: [0,(1252,[13,17,18,20,30,37,45,50,51,53,63,64,96,101,108,125,174,189,214,221,224,227,238,268,291,309,328,357,362,437,441,455,492,493,511,528,561,613,619,674,764,823,839,980,1098,1143],[1.0,1.0,2.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,1.0,3.0,1.0,2.0,1.0,5.0,1.0,2.0,2.0,1.0,4.0,1.0,2.0,3.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,1.0,1.0,1.0,1.0,2.0,1.0,2.0,1.0,1.0,1.0])] (of class org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema)
有一个LDA的工作样例,没有抛出任何问题。
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.Row
import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.mllib.clustering.{DistributedLDAModel, LDA}
val a = Vectors.dense(Array(1.0,2.0,3.0))
val b = Vectors.dense(Array(3.0,4.0,5.0))
val df = Seq((1L,a),(2L,b),(2L,a)).toDF
val ldaDF = df.map { case Row(id: Long, countVector: Vector) => (id, countVector) }
val model = new LDA().setK(3).run(ldaDF.javaRDD)
display(df)
唯一的区别在于第二个片段中我们使用的是密集矩阵。
Exception in thread "main" org.apache.spark.sql.AnalysisException: cannot resolve 'UDF(VecFunction)' due to data type mismatch: argument 1 requires vector type, however, 'VecFunction' is of vector type.;
请注意,参数和期望输入都被称为向量类型。 - Béatrice Moissinacorg.apache.spark.mllib.*linalg*.Vectors.fromML
,顺便说一下,那很有帮助 ;) - WestCoastProjects