526互联

spark dataset dataframe 动态添加列

发布时间 2023-07-31 17:28:07作者: iullor

需求

利用SparkSQL计算每一行数据的数据质量，如果数据不为NULL或者不为空字符串(或者符合正则表达式)，那么该字段该行数据积一分

网上解决方案

https://blog.csdn.net/Code_LT/article/details/87719115
https://blog.csdn.net/LLJJYY001/article/details/88964961?spm=1001.2101.3001.6650.1&utm_medium=distribute.pc_relevant.none-task-blog-2~default~CTRLIST~Rate-1-88964961-blog-87719115.235^v38^pc_relevant_yljh&depth_1-utm_source=distribute.pc_relevant.none-task-blog-2~default~CTRLIST~Rate-1-88964961-blog-87719115.235^v38^pc_relevant_yljh&utm_relevant_index=2

package com.emmm.test.scala

import org.apache.spark.SparkConf
import org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema
import org.apache.spark.sql.types.{StringType, StructType}
import org.apache.spark.sql.{Row, SparkSession}

object Emmm {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
    conf.setMaster("local[*]")
    conf.setAppName(this.getClass.getSimpleName)
    conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    conf.set("spark.kryo.registrationRequired", "true")
    conf.registerKryoClasses(Array(
      Class.forName("scala.collection.mutable.WrappedArray$ofRef"),
      Class.forName("org.apache.spark.sql.types.StringType$"),
      classOf[TPerson],
      classOf[org.apache.spark.sql.catalyst.expressions.GenericRowWithSchema],
      classOf[org.apache.spark.sql.types.StructType],
      classOf[org.apache.spark.sql.types.StructField],
      classOf[org.apache.spark.sql.types.Metadata],
      classOf[Array[TPerson]],
      classOf[Array[org.apache.spark.sql.Row]],
      classOf[Array[org.apache.spark.sql.types.StructField]],
      classOf[Array[Object]]
    ))
    val spark = SparkSession.builder()
      .config(conf)
      .getOrCreate()
    import spark.implicits._
    // 使用样例类创建RDD并转化成DF后又回到RDD
    spark.sparkContext.parallelize(Seq(TPerson("zs", "21"), TPerson("ls", "25"))).toDF().rdd
      .map(row => {
        // 打印schema
        println(row.schema)
        // 得到Row中的数据并往其中添加我们要新增的字段值
        val buffer = Row.unapplySeq(row).get.map(_.asInstanceOf[String]).toBuffer
        buffer.append("男") //增加一个性别
        buffer.append("北京") //增肌一个地址

        // 获取原来row中的schema,并在原来Row中的Schema上增加我们要增加的字段名以及类型.
        val schema: StructType = row.schema
          .add("gender", StringType)
          .add("address", StringType)
        // 使用Row的子类GenericRowWithSchema创建新的Row
        val newRow: Row = new GenericRowWithSchema(buffer.toArray, schema)
        // 使用新的Row替换成原来的Row
        newRow
      }).map(row => {
      // 打印新的schema
      println(row.schema)
      // 测试我们新增的字段
      val gender = row.getAs[String]("gender")
      // 获取原本就有的字段
      val name = row.getAs[String]("name")
      val age = row.getAs[String]("age")
      // 获取新的字段
      val address = row.getAs[String]("address")
      // 输出查看结果
      println(s"$name-$age-$gender-$address")
      row
    }).collect()
    spark.stop()
  }

  /**
    * 样例类
    *
    * @param name name属性
    * @param age  age属性
    */
  case class TPerson(name: String, age: String)

}

遇到问题

  • Row 行 Value列表 List 新增加一个元素，但是未生效，可能未考虑返回值 newList = List.append('new') 接收，详细查看一个scala List 集合返回值
  • freme.map(fun)(Encoder) 序列化问题，网上推荐一般使用 Encoders.kryo[] ，但是这个如果不传参数的话，默认返回值是一个BinaryType 而后row里面的返回值也变成byte二进制数组和实际需求渐行渐远，故不考虑

解决

部分核心代码，需要导入org.apache.spark.sql.catalyst.encoders.RowEncoder 序列化器，跟了一下SparkSQL的源码 ，底层使用的也是该类


    import org.apache.spark.sql.catalyst.encoders.RowEncoder
    val fields: Array[StructField] = frame.schema.fields
    val newFields = fields :+ StructField("score", IntegerType)
    val scoreDataset: Dataset[Row] = frame
      .map(row => {
        var score = 0
        val map: Map[String, Nothing] = row.getValuesMap(columns)
        map.foreach(m => {
          if (m._2 != null && m._2.toString.trim.nonEmpty) {
            score = score + 1
          }
        })
        // row.toSeq: _* 将原来row内的元素展开存放到新数组里面 .++(Array(score)) 两个数组拼接形成新的数组
        val array: Array[Any] = Array(row.toSeq: _*).++(Array(score))
        //重新创建一个Row对象，将数组中的元素 展开然后形成新的行value
        val newRow = Row(array: _*)
        //将新的row返回
        newRow
        //使用到了Spark同款序列化器，传入新增字段的 score 类型
      })(RowEncoder(StructType(newFields)))

参考