Как вставить элемент в массив rdd в искровом

Question

Привет Я попытался вставить элемент в массив rdd [String], используя scala в искры.

Вот пример.

val data = RDD[Array[String]] = Array(Array(1,2,3), Array(1,2,3,4), Array(1,2)). 

Я хочу сделать длину 4 из всех массивов в этих данных.

Если длина массива меньше 4, я хочу заполнить значение NULL в массиве.

вот мой код, который я пытался решить.

val newData = data.map(x => 
    if(x.length < 4){ 
     for(i <- x.length until 4){ 
     x.union("NULL") 
     } 
    } 
    else{ 
     x 
    } 
) 

Но результат Array[Any] = Array((), Array(1, 2, 3, 4),()).

Так что я попробовал другие способы. Я использовал yield для цикла.

val newData = data.map(x => 
    if(x.length < 4){ 
     for(i <- x.length until 4)yield{ 
     x.union("NULL") 
     } 
    } 
    else{ 
     x 
    } 
) 

В результате Array[Object] = Array(Vector(Array(1, 2, 3, N, U, L, L)), Array(1, 2, 3, 4), Vector(Array(1, 2, N, U, L, L), Array(1, 2, N, U, L, L)))

это не то, что я хочу. Я хочу, чтобы вернуться как этот

RDD[Array[String]] = Array(Array(1,2,3,NULL), Array(1,2,3,4), Array(1,2,NULL,NULL)).

Что я должен делать? Есть ли способ его решения?

Answer 1

union - функциональная операция, она не меняет массив x. Однако вам не нужно делать это с помощью цикла, и любые реализации цикла, вероятно, будут медленнее - гораздо лучше создать одну новую коллекцию со всеми значениями NULL, а не мутировать что-то каждый раз, когда вы добавляете нуль. Вот функция лямбда, которая должна работать для вас:

def fillNull(x: Array[Int], desiredLength: Int): Array[String] = { 
    x.map(_.toString) ++ Array.fill(desiredLength - x.length)("NULL") 
} 

val newData = data.map(fillNull(_, 4)) 

Answer 2

Я решил ваш случай использования со следующим кодом:

val initialRDD = sparkContext.parallelize(Array(Array[AnyVal](1, 2, 3), Array[AnyVal](1, 2, 3, 4), Array[AnyVal](1, 2, 3))) 
val transformedRDD = initialRDD.map(array => 
    if (array.length < 4) { 
    val transformedArray = Array.fill[AnyVal](4)("NULL") 
    Array.copy(array, 0, transformedArray, 0, array.length) 
    transformedArray 
    } else { 
    array 
    } 
) 
val result = transformedRDD.collect()