scala – 在Spark 2.0中访问向量列时出现MatchError

使用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应验证已删除的停用词

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.transform(filtered_df)
           .select("id","text","features","filtered").show()

在此之后,我看到在LDA中拟合此矢量化器的问题.我认为是CountVectorizer的问题是给出稀疏向量,但LDA需要密集向量.仍在努力找出问题所在.

以下是map无法转换的例外情况.

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,3.0,5.0,4.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,3.0))
val b = Vectors.dense(Array(3.0,5.0))
val df = Seq((1L,a),(2L,b),a)).toDF

val ldaDF = df.map { case Row(id: Long,countVector) } 

val model = new LDA().setK(3).run(ldaDF.javaRDD)
display(df)

唯一的区别是在第二个片段中我们有一个密集的矩阵.