MatchError при доступе к векторному столбцу в Spark 2.0

Я пытаюсь создать модель LDA в файле JSON.

Создание контекста искры с помощью файла JSON:

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)

Получить stopwords

%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

Собирая все stopwords

val stopwords = sc.textFile("/tmp/stopwords").collect()

Фильтрация stopwords

 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)

Разница лишь во втором фрагменте, у нас плотная матрица.

Источник

3 ответа

Решение

Это не имеет ничего общего с редкостью. Начиная с Spark 2.0.0 ML Transformers больше не генерировать o.a.s.mllib.linalg.VectorUDT но o.a.s.ml.linalg.VectorUDT и отображаются локально на подклассы o.a.s.ml.linalg.Vector, Они не совместимы со старым MLLib API, который постепенно устарел в Spark 2.0.0.

Вы можете конвертировать между "старый", используя Vectors.fromML:

import org.apache.spark.mllib.linalg.{Vectors => OldVectors}
import org.apache.spark.ml.linalg.{Vectors => NewVectors}

OldVectors.fromML(NewVectors.dense(1.0, 2.0, 3.0))
OldVectors.fromML(NewVectors.sparse(5, Seq(0 -> 1.0, 2 -> 2.0, 4 -> 3.0)))

но имеет смысл использовать ML реализация LDA, если вы уже используете трансформаторы ML.

Для удобства вы можете использовать неявные преобразования:

import scala.languageFeature.implicitConversions

object VectorConversions {
  import org.apache.spark.mllib.{linalg => mllib}
  import org.apache.spark.ml.{linalg => ml}

  implicit def toNewVector(v: mllib.Vector) = v.asML
  implicit def toOldVector(v: ml.Vector) = mllib.Vectors.fromML(v)
}
Источник

Я изменился:

val ldaDF = countVectors.map { 
             case Row(id: String, countVector: Vector) => (id, countVector) 
            }

чтобы:

val ldaDF = countVectors.map { case Row(docId: String, features: MLVector) => 
                               (docId.toLong, Vectors.fromML(features)) }

И это сработало как шарм! Это соответствует тому, что написал @zero323.

Список импорта:

import org.apache.spark.ml.feature.{CountVectorizer, RegexTokenizer, StopWordsRemover}
import org.apache.spark.ml.linalg.{Vector => MLVector}
import org.apache.spark.mllib.clustering.{LDA, OnlineLDAOptimizer}
import org.apache.spark.mllib.linalg.Vectors
import org.apache.spark.sql.{Row, SparkSession}
Источник

Решение очень простое, ребята.. найти ниже

//import org.apache.spark.mllib.linalg.Vector
import org.apache.spark.ml.linalg.Vector
Источник
Другие вопросы по тегам