Step 1: 思路

parse 方法应该做的事情应该有两件

• 捕获异常

  异常一定是要捕获的, 无论是否要抛给 DataFrame, 都要先捕获一下, 获知异常信息

  捕获要使用 try …? catch …? 代码块

• 返回结果

  返回结果应该分为两部分来进行说明

  • 正确, 正确则返回数据

  • 错误, 则应该返回两类信息, 一 告知外面哪个数据出了错, 二 告知错误是什么

对于这种情况, 可以使用 Scala 中提供的一个类似于其它语言中多返回值的 Either. Either 分为两个情况, 一个是 Left, 一个是 Right, 左右两个结果所代表的意思可有由用户来指定

val process = (b: Double) => {       
  val a = 10.0
  a / b
}

def safe(function: Double => Double, b: Double): Either[Double, (Double, Exception)] = {  
  try {
    val result = function(b)         
    Left(result)
  } catch {
    case e: Exception => Right(b, e) 
  }
}

val result = safe(process, 0)        

result match {                       
  case Left(r) => println(r)
  case Right((b, e)) => println(b, e)
}

 

	一个函数, 接收一个参数, 根据参数进行除法运算
	一个方法, 作用是让 process 函数调用起来更安全, 在其中 catch 错误, 报错后返回足够的信息 (报错时的参数和报错信息)
	正常时返回 Left, 放入正确结果
	异常时返回 Right, 放入报错时的参数, 和报错信息
	外部调用
	处理调用结果, 如果是 Right 的话, 则可以进行响应的异常处理和弥补

Either 和 Option 比较像, 都是返回不同的情况, 但是 Either 的 Right 可以返回多个值, 而 None 不行

如果一个 Either 有两个结果的可能性, 一个是 Left[L], 一个是 Right[R], 则 Either 的范型是 Either[L, R]

Step 2: 完成代码逻辑

加入一个 Safe 方法, 更安全

object TaxiAnalysisRunner {

  def main(args: Array[String]): Unit = {
    // ...

    // 4. 数据转换和清洗
    val taxiParsed = taxiRaw.rdd.map(safe(parse))
  }

  /**
    * 包裹转换逻辑, 并返回 Either
    */
  def safe[P, R](f: P => R): P => Either[R, (P, Exception)] = {
    new Function[P, Either[R, (P, Exception)]] with Serializable {
      override def apply(param: P): Either[R, (P, Exception)] = {
        try {
          Left(f(param))
        } catch {
          case e: Exception => Right((param, e))
        }
      }
    }
  }

  def parse(row: Row): Trip = {...}

  def parseTime(row: RichRow, field: String): Long = {...}

  def parseLocation(row: RichRow, field: String): Double = {...}
}

case class Trip(..)

class RichRow(row: Row) {...}

 

Step 3: 针对转换异常进行处理

对于 Either 来说, 可以获取 Left 中的数据, 也可以获取 Right 中的数据, 只不过如果当 Either 是一个 Right 实例时候, 获取 Left 的值会报错

所以, 针对于 Dataset[Either] 可以有如下步骤

1. 试运行, 观察是否报错

2. 如果报错, 则打印信息解决报错

3. 如果解决不了, 则通过 filter 过滤掉 Right

4. 如果没有报错, 则继续向下运行

object TaxiAnalysisRunner {

  def main(args: Array[String]): Unit = {
    ...

    // 4. 数据转换和清洗
    val taxiParsed = taxiRaw.rdd.map(safe(parse))
    val taxiGood = taxiParsed.map( either => either.left.get ).toDS()
  }

  ...
}

...

 

很幸运, 在运行上面的代码时, 没有报错, 如果报错的话, 可以使用如下代码进行过滤

object TaxiAnalysisRunner {

  def main(args: Array[String]): Unit = {
    ...

    // 4. 数据转换和清洗
    val taxiParsed = taxiRaw.rdd.map(safe(parse))
    val taxiGood = taxiParsed.filter( either => either.isLeft )
      .map( either => either.left.get )
      .toDS()
  }

  ...
}

...

Either。修改parse让其更安全：

/**
    * 作用就是封装 parse 方法, 捕获异常
    */
  //p是参数R是返回值
  def safe[P, R](f: P => R): P => Either[R, (P, Exception)] = {
    new Function[P, Either[R, (P, Exception)]] with Serializable {

      override def apply(param: P): Either[R, (P, Exception)] = {
        try {
          Left(f(param))
        } catch {
          case e: Exception => Right((param, e))
        }
      }
    }
  }

绘制直方图：

   // 5. 绘制时长直方图
    // 5.1 编写 UDF 完成时长计算, 将毫秒转为小时单位
    val hours = (pickUpTime: Long, dropOffTime: Long) => {
      val duration = dropOffTime - pickUpTime
      val hours = TimeUnit.HOURS.convert(duration, TimeUnit.MILLISECONDS)
      hours
    }
    val hoursUDF = udf(hours)
    // 5.2 进行统计
   taxiGood.groupBy(hoursUDF($"pickUpTime", $"dropOffTime") as "duration")
     .count()
     .sort("duration")
      .show()

减除反常数据：

    // 6. 根据直方图的显示, 查看数据分布后, 剪除反常数据
    spark.udf.register("hours", hours)
    val taxiClean = taxiGood.where("hours(pickUpTime, dropOffTIme) BETWEEN 0 AND 3")
    taxiClean.show()