如何向Spark RDD添加新列？

Question

我有一个RDD MANY列(例如,几百个),我怎么在这个RDD的末尾添加一个多列？

例如,如果我的RDD如下所示:

    123, 523, 534, ..., 893
    536, 98, 1623, ..., 98472
    537, 89, 83640, ..., 9265
    7297, 98364, 9, ..., 735
    ......
    29, 94, 956, ..., 758

如何向其中添加一列,其值是第二列和第三列的总和？

非常感谢你.

Answer 1

您根本不必使用Tuple*对象来添加新列RDD.

可以通过映射每一行,获取其原始内容以及要追加的元素来完成,例如:

val rdd = ...
val withAppendedColumnsRdd = rdd.map(row => {
  val originalColumns = row.toSeq.toList
  val secondColValue = originalColumns(1).asInstanceOf[Int]
  val thirdColValue = originalColumns(2).asInstanceOf[Int]
  val newColumnValue = secondColValue + thirdColValue 
  Row.fromSeq(originalColumns :+ newColumnValue)
  // Row.fromSeq(originalColumns ++ List(newColumnValue1, newColumnValue2, ...)) // or add several new columns
})

Answer 2

您有元组4的RDD，应用地图并将其转换为元组5

val rddTuple4RDD = ...........
val rddTuple5RDD = rddTuple4RDD.map(r=> Tuple5(rddTuple4._1, rddTuple4._2, rddTuple4._3, rddTuple4._4, rddTuple4._2 + rddTuple4._3))