Apache spark and scala, error while executing queries

I am working with a dataset whose sample is as follows:

"age";"job";"marital";"education";"default";"balance";"housing";"loan";"contact";"day";"month";"duration";"campaign";"pdays";"previous";"poutcome";"y"
58;"management";"married";"tertiary";"no";2143;"yes";"no";"unknown";5;"may";261;1;-1;0;"unknown";"no"
44;"technician";"single";"secondary";"no";29;"yes";"no";"unknown";5;"may";151;1;-1;0;"unknown";"no"

I have executed the following commands successfully:

import org.apache.spark.sql._
import org.apache.spark.sql.types._
import spark.sqlContext.implicits._
val data = sc.textFile(“file:///C:/Users/Desktop/bank-full-Copy.csv")
data.map(x => x.split(";(?=([^"]*"[^"]*")*[^"]*$)",-1))
val header = data.first()
val filtered = data.filter(x => x(0)!= header(0))
val rdds = filtered.map(x => Row(x(0).toInt,
x(1),
x(2),
x(3),
x(4),
x(5).toInt,
x(6),
x(7),
x(8),
x(9).toInt,
x(10),
x(11).toInt,
x(12).toInt,
x(13).toInt,
x(14).toInt,
x(15),
x(16) ))
val schema = StructType( List(StructField("age", IntegerType, true),
StructField("job", StringType, true) ,
StructField("marital", StringType, true),
StructField("education", StringType, true) ,
StructField("default", StringType, true),
StructField("balance", IntegerType, true) ,
StructField("housing", StringType, true) ,
StructField("loan", StringType, true) ,
StructField("contact", StringType, true) ,
StructField("day", IntegerType, true) ,
StructField("month", StringType, true) ,
StructField("duration", IntegerType, true) ,
StructField("campaign", IntegerType, true) ,
StructField("pdays", IntegerType, true) ,
StructField("previous", IntegerType, true) ,
StructField("poutcome", StringType, true) ,
StructField("y", StringType, true)) )
val df = spark.sqlContext.createDataFrame(rdds, schema)

I am getting following error:

df.groupBy("age","y").count.show()*,

java.lang.RuntimeException: Error while encoding: java.lang.RuntimeException: java.lang.Character is not a valid external type for schema of string

I am getting the same error when executing any query against the data. Can you please have a look and provide me with a solution?

Answer

If you looking to skip the RDD extra code you can use the below code

Input file csv (; delimited and every record separated by next line)

"age";"job";"marital";"education";"default";"balance";"housing";"loan";"contact";"day";"month";"duration";"campaign";"pdays";"previous";"poutcome";"y"
58;"management";"married";"tertiary";"no";2143;"yes";"no";"unknown";5;"may";261;1;-1;0;"unknown";"no"
44;"technician";"single";"secondary";"no";29;"yes";"no";"unknown";5;"may";151;1;-1;0;"unknown";"no"

• Define the struct schema
• Read ; delimited file
• Read the csv with header=true and pre defined schema as Dataframe directly

import org.apache.spark.sql.types.{IntegerType, StringType, StructField, StructType}

object ProcessSemiColonCsv {

  def main(args: Array[String]): Unit = {

    val spark = Constant.getSparkSess

    val schema = StructType( List(StructField("age", IntegerType, true),
      StructField("job", StringType, true) ,
      StructField("marital", StringType, true),
      StructField("education", StringType, true) ,
      StructField("default", StringType, true),
      StructField("balance", IntegerType, true) ,
      StructField("housing", StringType, true) ,
      StructField("loan", StringType, true) ,
      StructField("contact", StringType, true) ,
      StructField("day", IntegerType, true) ,
      StructField("month", StringType, true) ,
      StructField("duration", IntegerType, true) ,
      StructField("campaign", IntegerType, true) ,
      StructField("pdays", IntegerType, true) ,
      StructField("previous", IntegerType, true) ,
      StructField("poutcome", StringType, true) ,
      StructField("y", StringType, true)) )

    val df = spark.read
      .option("delimiter", ";")
      .option("header", "true")
      .schema(schema)
      .csv("src/main/resources/SemiColon.csv")

    df.show()
    df.printSchema()
  }

}