如何将元素插入到spark中的rdd数组答案

【问题标题】：how to insert element to rdd array in spark如何将元素插入到spark中的rdd数组
【发布时间】：2017-04-19 17:30:36
【问题描述】：

您好，我尝试在 spark 中使用 scala 将元素插入 rdd 数组[String]。

这里是例子。

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), ())。

所以我尝试了另一种方法。我在 for 循环中使用了 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)).

我该怎么办？有办法解决吗？

【问题讨论】：

标签： arrays scala apache-spark rdd

【解决方案1】：

我用以下代码解决了你的用例：

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()

【讨论】：

这个硬编码的4 让我觉得我可能使用了array.length 上限的广播。
非常感谢！很有用！

【解决方案2】：

union 是一个函数操作，它不会改变数组x。但是，您不需要使用循环来执行此操作，并且任何循环实现可能会更慢 - 使用所有 NULL 值创建一个新集合要好得多，而不是每次添加 null 时都改变一些东西。这是一个适合您的 lambda 函数：

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

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

【讨论】：

感谢您的建议！非常感谢！我试试看！