Spark-Scala：使用异常处理将固定宽度线解析为 Dataframe Api

Question

我是初学者使用 scala 学习 spark .pardon 为我的英语不好...我需要编写一个程序来使用 spark-scala Dataframe Api 将分隔和固定宽度的文件解析为 Dataframe。此外，如果输入数据已损坏，则程序必须按以下给定方式处理：

A:ignoring the input data
B:investigate the error in input
C:stop on error

为了实现上述目标，我已经使用 DataFrame Api 选项成功完成了对分隔文件的异常处理解析。但我不知道如何对固定宽度文件应用相同的技术。我使用的是 Spark 2.4.3 版本。

// predefined schema used in program
val schema = new StructType()
.add("empno",IntegerType,true)
.add("ename",StringType,true)
.add("designation",StringType,true)
.add("manager",StringType,true)
.add("hire_date",StringType,true)
.add("salary",DoubleType,true)
.add("deptno",IntegerType,true)
.add("_corrupt_record", StringType, true)

// parse csv file into DataFrame Api
// option("mode","PERMISSIVE") used to handle corrupt record
val textDF =sqlContext.read.format("csv").option("header", "true").schema(schema).option("mode", "PERMISSIVE").load("empdata.csv")
textDF.show

// program for fixed width line

// created lsplit method to split line into list of tokens based on width input / string

def lsplit(pos: List[Int], str: String): List[String] = {
val (rest, result) = pos.foldLeft((str, List[String]())) {
case ((s, res),curr) =>
    if(s.length()<=curr)
    {
    val split=s.substring(0).trim()
    val rest=""
    (rest, split :: res)
    }
    else if(s.length()>curr)
    {
    val split=s.substring(0, curr).trim()
    val rest=s.substring(curr)
    (rest, split :: res)
    }
    else
    {
    val split=""
    val rest=""
    (rest, split :: res)
    }
}
// list is reversed
result.reverse
}
// create case class to hold parsed data
case class EMP(empno:Int,ename:String,designation:String,manager:String,hire_dt:String,salary:Double,deptno:Int)


// create variable to hold width length
val sizeOfColumn=List(4,4,5,4,10,8,2);

// code to transform string to case class record
val ttRdd=textDF.map { 
    x => 
    val row=lsplit(sizeOfColumn,x.mkString) 
    EMP(row(0).toInt,row(1),row(2),row(3),row(4).toDouble,row(5).toInt)
}


Code works fine for proper data but fails if incorrect data comes in file.
for e.g: "empno" column has some non-integer data..program throws exception NumberFormatException..
The program must handle if actual data in file does not match the specified schema as handled in delimited file.

请在这里帮助我。我需要对固定宽度文件使用与分隔文件相同的方法。

Answer 1

这很明显。

您正在将自己的方法与 API“许可”选项相结合。

permissive 会拾取错误数据类型等错误。然后你自己的进程 lsplit 仍然执行并且可以得到一个空异常。例如。如果我输入 empnum "YYY" 这显然是可以观察到的。

如果数据类型OK，长度错误，大部分情况下你处理正确，但是字段是乱码。

您的 lsplit 需要更加健壮，并且您需要检查其中是否存在错误或在调用不调用之前是否存在错误。

第一种情况

+-----+-----+---------------+
|empno|ename|_corrupt_record|
+-----+-----+---------------+
| null| null|      YYY,Gerry|
| 5555|Wayne|           null|
+-----+-----+---------------+

org.apache.spark.SparkException: Job aborted due to stage failure: Task 0 in stage 30.0 failed 1 times, most recent failure: Lost task 0.0 in stage 30.0 (TID 30, localhost, executor driver): java.lang.NumberFormatException: For input string: "null"

第二种情况

+------+-----+---------------+
| empno|ename|_corrupt_record|
+------+-----+---------------+
|444444|Gerry|           null|
|  5555|Wayne|           null|
+------+-----+---------------+

res37: Array[EMP] = Array(EMP(4444,44Ger), EMP(5555,Wayne))

简而言之，有些工作要做，实际上不需要标题。