如何更新RDD [Row]中的一个列值？

1 回答

• 2

  您可以定义这样的 update 方法来更新Row中的字段：

  import org.apache.spark.sql.Row
  
  def update(r: Row): Row = {
      val s = r.toSeq
      Row.fromSeq((s.take(4) :+ s(4).asInstanceOf[String].take(10)) ++ s.drop(5))
  }
  
  rdd.map(update(_)).collect
  
  //res13: Array[org.apache.spark.sql.Row] = 
  //       Array([510116,8042,1,8298,2017090718,1,3,lineno805], 
  //             [510116,8042,1,8152,2017090718,1,3,lineno805], 
  //             [510116,8042,1,8154,2017090716,1,3,lineno805], 
  //             [510116,8042,1,8069,2017090716,1,3,lineno805], 
  //             [510116,8042,1,8061,2017090717,1,3,lineno805], 
  //             [510116,8042,1,9906,2017090717,1,3,lineno805], 
  //             [510116,8042,1,8295,2017090717,1,3,lineno805])
  

  ---

  一种更简单的方法是使用DataFrame API和 substring 函数：

  1）从rdd创建一个数据框：

  val df = spark.createDataFrame(rdd, rdd.take(1)(0).schema)
  // df: org.apache.spark.sql.DataFrame = [_c0: string, _c1: string ... 6 more fields]
  

  2）使用 substring 转换列：

  df.withColumn("_c4", substring($"_c4", 0, 10)).show
  +------+----+---+----+----------+---+---+---------+
  |   _c0| _c1|_c2| _c3|       _c4|_c5|_c6|      _c7|
  +------+----+---+----+----------+---+---+---------+
  |510116|8042|  1|8298|2017090718|  1|  3|lineno805|
  |510116|8042|  1|8152|2017090718|  1|  3|lineno805|
  |510116|8042|  1|8154|2017090716|  1|  3|lineno805|
  |510116|8042|  1|8069|2017090716|  1|  3|lineno805|
  |510116|8042|  1|8061|2017090717|  1|  3|lineno805|
  |510116|8042|  1|9906|2017090717|  1|  3|lineno805|
  |510116|8042|  1|8295|2017090717|  1|  3|lineno805|
  +------+----+---+----+----------+---+---+---------+
  

  3）将数据帧转换为rdd很容易：

  val getSerialRdd = df.withColumn("_c4", substring($"_c4", 0, 10)).rdd
  

  回复于 2024-04-29T07:42:05+08:00