Spark Scala 将 Dataframe 空白记录替换为“0”

Question

我需要将我的 Dataframe 字段的空白记录替换为“0”

这是我的代码 -->

import sqlContext.implicits._

case class CInspections (business_id:Int, score:String, date:String, type1:String)

val baseDir = "/FileStore/tables/484qrxx21488929011080/"
val raw_inspections = sc.textFile (s"$baseDir/inspections_plus.txt")
val raw_inspectionsmap = raw_inspections.map ( line => line.split ("\t"))
val raw_inspectionsRDD = raw_inspectionsmap.map ( raw_inspections => CInspections (raw_inspections(0).toInt,raw_inspections(1), raw_inspections(2),raw_inspections(3)))
val raw_inspectionsDF = raw_inspectionsRDD.toDF
raw_inspectionsDF.createOrReplaceTempView ("Inspections")
raw_inspectionsDF.printSchema
raw_inspectionsDF.show()

我正在使用案例类，然后转换为 Dataframe。但我需要“分数”作为 Int，因为我必须执行一些操作并对其进行排序。 但是，如果我将其声明为 score:Int，那么我会收到空​​白值错误。

java.lang.NumberFormatException：对于输入字符串：“”

+-----------+-----+--------+--------------------+
|business_id|score|    date|               type1|
+-----------+-----+--------+--------------------+
|         10|     |20140807|Reinspection/Foll...|
|         10|   94|20140729|Routine - Unsched...|
|         10|     |20140124|Reinspection/Foll...|
|         10|   92|20140114|Routine - Unsched...|
|         10|   98|20121114|Routine - Unsched...|
|         10|     |20120920|Reinspection/Foll...|
|         17|     |20140425|Reinspection/Foll...|
+-----------+-----+--------+--------------------+

我需要将 score 字段作为 Int，因为对于以下查询，它排序为 String 而不是 Int 并给出错误结果

sqlContext.sql("""select raw_inspectionsDF.score  from raw_inspectionsDF where score <>"" order by score""").show()

+-----+
|score|
+-----+
|  100|
|  100|
|  100|
+-----+

Answer 1

空字符串无法转成Integer，需要将Score设为nullable，这样如果字段缺失，则表示为null，可以尝试以下方法：

import scala.util.{Try, Success, Failure}

1) 定义一个自定义的解析函数，如果字符串不能转换为 Int，则返回 None，在你的情况下为空字符串；

def parseScore(s: String): Option[Int] = {
  Try(s.toInt) match {
    case Success(x) => Some(x)
    case Failure(x) => None
  }
}

2) 将案例类中的 score 字段定义为Option[Int] 类型；

case class CInspections (business_id:Int, score: Option[Int], date:String, type1:String)

val raw_inspections = sc.textFile("test.csv")
val raw_inspectionsmap = raw_inspections.map(line => line.split("\t"))

3）使用自定义的parseScore函数解析score字段；

val raw_inspectionsRDD = raw_inspectionsmap.map(raw_inspections => 
    CInspections(raw_inspections(0).toInt, parseScore(raw_inspections(1)), 
                 raw_inspections(2),raw_inspections(3)))

val raw_inspectionsDF = raw_inspectionsRDD.toDF
raw_inspectionsDF.createOrReplaceTempView ("Inspections")

raw_inspectionsDF.printSchema
//root
// |-- business_id: integer (nullable = false)
// |-- score: integer (nullable = true)
// |-- date: string (nullable = true)
// |-- type1: string (nullable = true)

raw_inspectionsDF.show()

+-----------+-----+----+-----+
|business_id|score|date|type1|
+-----------+-----+----+-----+
|          1| null|   a|    b|
|          2|    3|   s|    k|
+-----------+-----+----+-----+

4) 正确解析文件后，可以使用na函数fill轻松将空值替换为0：

raw_inspectionsDF.na.fill(0).show
+-----------+-----+----+-----+
|business_id|score|date|type1|
+-----------+-----+----+-----+
|          1|    0|   a|    b|
|          2|    3|   s|    k|
+-----------+-----+----+-----+