SPARK中管道分隔文件中的数据帧

Question

我是SPARK的新手，所以尝试做一个小程序，遇到了下面的错误。有人能帮帮忙吗?

仅供参考-该程序似乎在示例文件中的列中没有空数据时工作，但问题似乎是由于第二行中的空值。

数据:内容TEMP_EMP.dat

1232|JOHN|30|IT
1532|DAVE|50|
1542|JEN|25|QA

SCALA代码将这些数据解析为dataframe

import org.apache.spark.sql.Row;
import org.apache.spark.sql.types.{StructType, StructField, StringType};
val employee = sc.textFile("file:///TEMP_EMP.dat")
val textFileTemp = sc.textFile("file:///TEMP_EMP.dat");
val schemaString = "ID|NAME|AGE|DEPT";
val schema = StructType(schemaString.split('|').map(fieldName=>StructField(fieldName,StringType,true)));
val rowRDD = employee.map(_.split('|')).map(e => Row(e(0),e(1),e(2), e(3) ));
val employeeDF = sqlContext.createDataFrame(rowRDD, schema);
employeeDF.registerTempTable("employee");
val allrecords = sqlContext.sql("SELECT * FROM employee");
allrecords.show();

错误日志:

WARN 2016-08-17 13:36:21 006 org.apache.spark.scheduler.TaskSetManager: Lost task 0.0 in stage 6.0: java.lang.ArrayIndexOutOfBoundsException: 3

Answer 1

这一行:

val rowRDD = employee.map(_.split('|')).map(e => Row(e(0),e(1),e(2), e(3) ));

假设employee.map(_.split('|'))的结果至少有4个元素，但第二行只有3个元素，因此出现了索引越界异常。

游戏介绍:

scala> val oneRow = "1532|DAVE|50|".split('|')
oneRow: Array[String] = Array(1532, DAVE, 50)
scala> oneRow(3)
java.lang.ArrayIndexOutOfBoundsException: 3

Answer 2

我们应该这样分割它:

val schema = StructType(
                schemaString
                   .split("|",-1)
                   .map( fieldName => StructField(fieldName,StringType,true) )
             );
val rowRDD = employee
                .map( _.split("|", -1) )
                .map( e => Row(e(0),e(1),e(2),e(3)) );