Spark Scala для чтения текстового файла с картой и фильтром

Question

У меня есть текстовый файл в следующем формате (id,f1,f2,f3,...,fn):

12345,0,0,1,2,...,3
23456,0,0,1,2,...,0
33333,0,1,1,0,...,0
56789,1,0,0,0,...,4
a_123,0,0,0,6,...,3

И я хочу прочитать файл (игнорируйте строку типа a_123,0,0,0,6,...,3), чтобы создать RDD[(Long, Vector).Вот мое решение:

  def readDataset(path: String, sparkSession: SparkSession): RDD[(ItemId, Vector)] = {
    val sc = sparkSession.sparkContext
    sc.textFile(path)
      .map({ line => val values=line.split(",")
        (
          values(0).toLong,
          //util.Try(values(0).toLong).getOrElse(0L),
          Vectors.dense(values.slice(1, values.length).map {x => x.toDouble }).toSparse
        )})
      .filter(x => x._1 > 0)
  }

Однако этот код не может быть скомпилирован:

[ERROR]  found   : org.apache.spark.rdd.RDD[(Long, org.apache.spark.ml.linalg.SparseVector)]
[ERROR]  required: org.apache.spark.rdd.RDD[(Long, org.apache.spark.ml.linalg.Vector)]
[ERROR]     (which expands to)  org.apache.spark.rdd.RDD[(Long, org.apache.spark.ml.linalg.Vector)]
[ERROR] Note: (Long, org.apache.spark.ml.linalg.SparseVector) <: (Long, org.apache.spark.ml.linalg.Vector), but class RDD is invariant in type T.
[ERROR] You may wish to define T as +T instead. (SLS 4.5)
[ERROR]       .filter(x => x._1 > 0)
[ERROR]              ^
[ERROR] one error found

Но если я удалю . toSparse или .filter(x => x._1 > 0), этот код может быть успешно скомпилирован.

Кто-нибудь знает, почему и что я должен сделать, чтобы это исправить?

Также есть ли лучший способ прочитать файл в RDD с игнорированием нечисловых строк идентификатора?

Answer 1

Код успешно компилируется, если вы удалите toSparse, потому что тип вашего PairRDD - (ItemId, Vector).

Класс / тип org.apache.spark.ml.linalg.Vector представляет собой плотный вектор, который вы генерируете, используя Vector.dense, и когда вы вызываете toSparse, он преобразуется в org.apache.spark.ml.linalg.SparseVector, который не является типом, ожидаемым вашим PairRDD.

Что касается фильтрации нецелых идентификаторов, я бы сказал, что ваш метод - хороший способ сделать это.