什么是缺失值

一个值本身的含义是这个值不存在则称之为缺失值, 也就是说这个值本身代表着缺失, 或者这个值本身无意义, 比如说 null, 比如说空字符串

关于数据的分析其实就是统计分析的概念, 如果这样的话, 当数据集中存在缺失值, 则无法进行统计和分析, 对很多操作都有影响

缺失值如何产生的

Spark 大多时候处理的数据来自于业务系统中, 业务系统中可能会因为各种原因, 产生一些异常的数据

例如说因为前后端的判断失误, 提交了一些非法参数. 再例如说因为业务系统修改 MySQL 表结构产生的一些空值数据等. 总之在业务系统中出现缺失值其实是非常常见的一件事, 所以大数据系统就一定要考虑这件事.

缺失值的类型

常见的缺失值有两种

• null, NaN 等特殊类型的值, 某些语言中 null 可以理解是一个对象, 但是代表没有对象, NaN 是一个数字, 可以代表不是数字

  针对这一类的缺失值, Spark 提供了一个名为 DataFrameNaFunctions 特殊类型来操作和处理

• "Null", "NA", " " 等解析为字符串的类型, 但是其实并不是常规字符串数据

  针对这类字符串, 需要对数据集进行采样, 观察异常数据, 总结经验, 各个击破

  DataFrameNaFunctions

  当数据集中出现缺失值的时候, 大致有两种处理方式, 一个是丢弃, 一个是替换为某值, DataFrameNaFunctions 中包含一系列针对空值数据的方案

  • DataFrameNaFunctions.drop 可以在当某行中包含 null 或 NaN 的时候丢弃此行

  • DataFrameNaFunctions.fill 可以在将 null 和 NaN 充为其它值

  • DataFrameNaFunctions.replace 可以把 null 或 NaN 替换为其它值, 但是和 fill 略有一些不同, 这个方法针对值来进行替换

  •  // 1. 创建 SparkSession
      val spark = SparkSession.builder()
        .master("local[6]")
        .appName("null processor")
        .getOrCreate()
    
      @Test
      def nullAndNaN(): Unit = {
    
    
        // 2. 导入数据集
    
        // 3. 读取数据集
        //    1. 通过Saprk-csv自动的推断类型来读取, 推断数字的时候会将 NaN 推断为 字符串
    //    spark.read
    //      .option("header", true)
    //      .option("inferSchema", true)//这里是推断
    //      .csv(...)
        //    2. 直接读取字符串, 在后续的操作中使用 map 算子转类型
    //    spark.read.csv().map( row => row... )
        //    3. 指定 Schema, 不要自动推断
        val schema = StructType(
          List(
            StructField("id", LongType),
            StructField("year", IntegerType),
            StructField("month", IntegerType),
            StructField("day", IntegerType),
            StructField("hour", IntegerType),
            StructField("season", IntegerType),
            StructField("pm", DoubleType)
          )
        )
    
        val sourceDF = spark.read
          .option("header", value = true)
          .schema(schema)
          .csv("dataset/beijingpm_with_nan.csv")
    
        sourceDF.show()
    
        // 4. 丢弃
        // 2019, 12, 12, NaN
        // 规则:
        //      1. any, 只有有一个 NaN 就丢弃
        sourceDF.na.drop("any").show()
        sourceDF.na.drop().show()
        //      2. all, 所有数据都是 NaN 的行才丢弃
        sourceDF.na.drop("all").show()
        //      3. 某些列的规则
        sourceDF.na.drop("any", List("year", "month", "day", "hour")).show()
    
        // 5. 填充
        // 规则:
        //     1. 针对所有列数据进行默认值填充
        sourceDF.na.fill(0).show()
        //     2. 针对特定列填充
        sourceDF.na.fill(0, List("year", "month")).show()
      }

  

   

   

   

   

   

   

   

   

   

   

   

   遇到字符串的时候：

   

    @Test
    def strProcessor(): Unit = {
      // 读取数据集
      val sourceDF = spark.read
        .option("header", value = true)
        .option("inferSchema", value = true)
        .csv("dataset/BeijingPM20100101_20151231.csv")
  
  //    sourceDF.show()
  
      // 1. 丢弃
      import spark.implicits._
  //    sourceDF.where(‘PM_Dongsi =!= "NA").show()
  
      // 2. 替换
      import org.apache.spark.sql.functions._
      // select name, age, case
      // when ... then ...
      // when ... then ...
      // else
      sourceDF.select(
        ‘No as "id", ‘year, ‘month, ‘day, ‘hour, ‘season,
        when(‘PM_Dongsi === "NA", Double.NaN)//当啥的时候进行替换
          .otherwise(‘PM_Dongsi cast DoubleType)//转换类型
          .as("pm")
      ).show()
  
      // 原类型和转换过后的类型, 必须一致
      sourceDF.na.replace("PM_Dongsi", Map("NA" -> "NaN", "NULL" -> "null")).show()
      //指定哪一行进行怎样的替换
    }