Transform RDD в RowMatrix для PCA

Question

Исходные данные у меня есть выглядит следующим образом:

РДД данных:

key -> index

1 -> 2

1 -> 3

1 -> 5

2 -> 1

2 -> 3

2 -> 4

Как я могу преобразовать RDD в следующем формате?

key -> index1, index2, index3, index4, index5

1 -> 0,1,1,0,1

2 -> 1,0,1,1,0

Мой текущий метод:

val vectors = filtered_data_by_key.map(x => { 
    var temp = Array[AnyVal]() 
    x._2.copyToArray(temp) 
    (x._1, Vectors.sparse(filtered_key_size, temp.map(_.asInstanceOf[Int]), Array.fill(filtered_key_size)(1))) 
}) 

я получил какую-то странную ошибку:

org.apache.spark.SparkException: Job aborted due to stage failure: Task 3 in stage 54.0 failed 1 times, most recent failure: Lost task 3.0 in stage 54.0 (TID 75, localhost): java.lang.IllegalArgumentException: requirement failed

Когда я пытаюсь отлаживать эту программу, используя следующий код:

val vectors = filtered_data_by_key.map(x => { 
    val temp = Array[AnyVal]() 
    val t = x._2.copyToArray(temp) 
    (x._1, temp) 
}) 

Я нашел, что temp пуст, поэтому проблема находится в copyToArray().

Я не уверен, как это решить.

Answer 1

Я не совсем понял вопрос. Почему ваши ключи важны? И каково максимальное значение индекса? В вашем коде вы используете определенное количество ключей как максимальное значение индекса, но я считаю, что это ошибка.

Но я предполагаю, что максимальное значение индекса равно 5. В этом случае, я считаю, что это было бы то, что вы ищете:

val vectors = data_by_key.map({case(k,it)=>Vectors.sparse(5,it.map(x=>x-1).toArray, 
     Array.fill(it.size)(1))}) 

val rm = new RowMatrix(vectors) 

я уменьшил индексный номер один, потому что они должны начинаться с 0 .

Ошибка запроса «не удалась» из-за ваших векторов индекса и значений, не имеющих одинакового размера.