使用火花數據框進行字段數據驗證

Question

我有一堆列，樣本如我的數據顯示如下所示。 我需要檢查列的錯誤，並將不得不生成兩個輸出文件。 我正在使用Apache Spark 2.0，我希望以有效的方式做到這一點。

Schema Details 
--------------- 
EMPID - (NUMBER) 
ENAME - (STRING,SIZE(50)) 
GENDER - (STRING,SIZE(1)) 

Data 
---- 
EMPID,ENAME,GENDER 
1001,RIO,M 
1010,RICK,MM 
1015,123MYA,F 

我excepected輸出文件應該如圖6-8所示：

1. 
EMPID,ENAME,GENDER 
1001,RIO,M 
1010,RICK,NULL 
1015,NULL,F 

2. 
EMPID,ERROR_COLUMN,ERROR_VALUE,ERROR_DESCRIPTION 
1010,GENDER,"MM","OVERSIZED" 
1010,GENDER,"MM","VALUE INVALID FOR GENDER" 
1015,ENAME,"123MYA","NAME SHOULD BE A STRING" 

感謝

Answer 1

我還沒有真正與星火2.0工作，所以我會盡力與溶液回答你的問題Spark 1.6。

// Load you base data 
val input = <<you input dataframe>> 

//Extract the schema of your base data 
val originalSchema = input.schema 

// Modify you existing schema with you additional metadata fields 
val modifiedSchema= originalSchema.add("ERROR_COLUMN", StringType, true) 
            .add("ERROR_VALUE", StringType, true) 
            .add("ERROR_DESCRIPTION", StringType, true) 

// write a custom validation function         
def validateColumns(row: Row): Row = { 

var err_col: String = null 
var err_val: String = null 
var err_desc: String = null 
val empId = row.getAs[String]("EMPID") 
val ename = row.getAs[String]("ENAME") 
val gender = row.getAs[String]("GENDER") 

// do checking here and populate (err_col,err_val,err_desc) with values if applicable 

Row.merge(row, Row(err_col),Row(err_val),Row(err_desc)) 
} 

// Call you custom validation function 
val validateDF = input.map { row => validateColumns(row) } 

// Reconstruct the DataFrame with additional columns      
val checkedDf = sqlContext.createDataFrame(validateDF, newSchema) 

// Filter out row having errors 
val errorDf = checkedDf.filter($"ERROR_COLUMN".isNotNull && $"ERROR_VALUE".isNotNull && $"ERROR_DESCRIPTION".isNotNull) 

// Filter our row having no errors 
val errorFreeDf = checkedDf.filter($"ERROR_COLUMN".isNull && !$"ERROR_VALUE".isNull && !$"ERROR_DESCRIPTION".isNull) 

我已經親自使用這種方法，它適用於我。我希望它能指引你朝着正確的方向前進。