无法使用 CSV 文件的 spark scala 显示数据帧记录

Question

我通过使用 map 函数将 RDD 转换为 DF 创建了 dataframe。当我尝试显示记录时，它给了我exception。
以下是我的代码：

//Created case class
case class employees(emp_id:java.lang.Long,emp_name:String, job_name:String,manager_id:java.lang.Long,hire_date:String,salary:java.lang.Double,commision:java.lang.Double,dep_id:java.lang.Long);

// Created DF
val employeesDf=rd1.map(_.split(",")).map(p=>employees(p(0).toLong,p(1),p(2),p(3).toLong,p(4),p(5).toDouble,p(6).toDouble,p(7).toLong)).toDF()

//Schema    
scala> employeesDf
    res5: org.apache.spark.sql.DataFrame = [emp_id: bigint, emp_name: string, job_name: string, manager_id: bigint, hire_date: string, salary: double, commision: double, dep_id: bigint]

但是当我尝试显示一些记录时，它会引发异常。下面是error：

scala> employeesDf.show()
18/08/05 07:08:43 ERROR executor.Executor: Exception in task 0.0 in stage 1.0 (TID 1)
java.lang.NumberFormatException: For input string: ""

以下是员工数据集：

DATASET IMAGE

那么我哪里错了？？我从几个小时就被困住了..

Answer 1

我通过创建 UDF 并在地图函数中使用它解决了我的问题。下面是代码：

//Create case class for schema :

case class employees(emp_id:java.lang.Long,emp_name:String, job_name:String,manager_id:java.lang.Long,hire_date:String,salary:java.lang.Double,commision:java.lang.Double,dep_id:java.lang.Long);

// Create UDF’s for Long and double :

def getDoubleValue(value:String):Double= {
  val output:Double=if (value != null && value.trim.length>0) {
    value.toDouble
  }else{
    0D
  }
  output
}
def getLongValue(value:String):Long= {
  val output:Long=if (value != null && value.trim.length>0) {
    value.toLong
  }else{
    0L
  }
  output
}

// Create RDD

val rdd=sc.textFile("file:////home/hduser/Desktop/Employees/employees.txt").filter(p=>{p!=null && p.trim.length>0})


// Create DF

val df=rdd.map(_.split(",")).map(p=>employees(getLongValue(p(0)),p(1),p(2),getLongValue(p(3)),p(4),getDoubleValue(p(5)),getDoubleValue(p(6)),getLongValue(p(7)))).toDF()

// Display records:

df.show();

Answer 2

您的dataframe's 某些列包含empty strings 并且您正在尝试将parsing 他们转为double、long。因此，要么将这些column 更改为strings，要么根据您的业务需求在解析中使用if-else 条件。如下所示

//Sample data in test.txt

65646,JONAS,MANAGER,68319,1991-04-02,2957.00,,2001
64989,ADELYN,SALESMAN,66928,1991-02-20,1700.00,400.00,3001

val rdd=sc.textFile("C:\\spark\\programs\\test.txt").filter(p=>{p!=null && p.trim.length>0})

// Created DF
rdd.map(_.split(",")).map(p=>employees(p(0).toLong,p(1),p(2),if(p(3).length>0) p(3).toLong else 0,p(4),if(p(5).length>0) p(5).toDouble else 0, if(p(6).length>0) p(6).toDouble else 0,p(7).toLong)).toDF().show

//Output: 

+------+--------+--------+----------+----------+------+---------+------+
|emp_id|emp_name|job_name|manager_id| hire_date|salary|commision|dep_id|
+------+--------+--------+----------+----------+------+---------+------+
| 65646|   JONAS| MANAGER|     68319|1991-04-02|2957.0|      0.0|  2001|
| 64989|  ADELYN|SALESMAN|     66928|1991-02-20|1700.0|    400.0|  2001|
+------+--------+--------+----------+----------+------+---------+------+