微信公众号搜"智元新知"关注
微信扫一扫可直接关注哦!

spark dataframe新增一列的四种方法

dataframe新增一列有如下四种常用方法
方法一:利用createDataFrame方法,新增列的过程包含在构建rdd和schema中
方法二:利用withColumn方法,新增列的过程包含在udf函数
方法三:利用sql代码,新增列的过程直接写入sql代码
方法四:以上三种是增加一个有判断的列,如果想要增加一列唯一序号,可以使用monotonically_increasing_id

//dataframe新增一列方法1,利用createDataFrame方法
val trdd = input.select(targetColumns).rdd.map(x=>{
  if (x.get(0).toString().todouble > critValueR || x.get(0).toString().todouble < critValueL) 
    Row(x.get(0).toString().todouble,"F")
  else Row(x.get(0).toString().todouble,"T")      
  })      
val schema = input.select(targetColumns).schema.add("flag", StringType, true)
val sample3 = ss.createDataFrame(trdd, schema).distinct().withColumnRenamed(targetColumns, "idx")

//dataframe新增一列方法2
val code :(Int => String) = (arg: Int) => {if (arg > critValueR || arg < critValueL) "F" else "T"}
val addCol = udf(code)
val sample3 = input.select(targetColumns).withColumn("flag", addCol(input(targetColumns)))
.withColumnRenamed(targetColumns, "idx")

//dataframe新增一列方法3
input.select(targetColumns).createOrReplaceTempView("tmp")
val sample3 = ss.sqlContext.sql("select distinct "+targetColname+
    " as idx,case when "+targetColname+">"+critValueR+" then 'F'"+
    " when "+targetColname+"<"+critValueL+" then 'F' else 'T' end as flag from tmp")

//添加序号列
import org.apache.spark.sql.functions.monotonically_increasing_id
val inputnew = input.withColumn("idx", monotonically_increasing_id)

注:内容来源于网络

版权声明:本文内容由互联网用户自发贡献,该文观点与技术仅代表作者本人。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如发现本站有涉嫌侵权/违法违规的内容, 请发送邮件至 [email protected] 举报,一经查实,本站将立刻删除。

相关推荐