Проверка полевых данных с использованием искрового фрейма данных

У меня есть куча столбцов, например, мои данные отображаются, как показано ниже. Мне нужно проверить столбцы на наличие ошибок и сгенерировать два выходных файла. Я использую 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

Мои ожидаемые выходные файлы должны быть такими, как показано ниже:

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"

Спасибо


apache-spark apache-spark-sql apache-spark-dataset
person 1pluszara    schedule 10.09.2017    source источник

Ответы (1)


arrow_upward
7
arrow_downward

Я толком не работал со Spark 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)

Я использовал этот подход лично, и он работает для меня. Я надеюсь, что это указывает вам в правильном направлении.

person Nilanjan Sarkar    schedule 10.09.2017
comment
Если это работает для вас, не могли бы вы принять ответ? А пока я подумаю над вашей проблемой, если я смогу что-нибудь придумать - person Nilanjan Sarkar; 14.09.2017