データフレームを作成中にscala.MatchError：1201（クラスjava.lang.Integerの）に直面する

Question

次のコードを実行して、テキストファイルからデータフレームを作成しています。

import org.apache.spark.SparkContext 
import org.apache.spark.SparkConf 
import org.apache.spark.sql.{SQLContext, Row} 
import org.apache.spark.sql.types.{StructType, StringType, StructField} 


/** 
    * Created by PSwain on 6/19/2016. 
    */ 
object RddToDataframe extends App { 

    val scnf=new SparkConf().setAppName("RddToDataFrame").setMaster("local[1]") 
    val sc = new SparkContext(scnf) 
    val sqlContext = new SQLContext(sc) 

    val employeeRdd=sc.textFile("C:\\Users\\pswain\\IdeaProjects\\test1\\src\\main\\resources\\employee") 

    //Creating schema 

    val employeeSchemaString="id name age" 
    val schema = StructType(employeeSchemaString.split(",").map(colNmae => StructField(colNmae,StringType,true))) 

    //Creating RowRdd 
    val rowRdd= employeeRdd.map(row => row.split(",")).map(row => Row(row(0).trim.toInt,row(1),row(2).trim.toInt)) 

    //Creating dataframe = RDD[rowRdd] + schema 
    val employeeDF=sqlContext.createDataFrame(rowRdd,schema). registerTempTable("Employee") 

    sqlContext.sql("select * from Employee").show() 


} 

InteliJで実行している間、私は以下のようにタイプの不一致エラーが見つかりました。このエラーがなぜ起こっているのかを特定できません。文字列を整数に変換するだけです。従業員ファイルは以下のように入力されていますが、すべて1行で表示されていますが、それぞれ1行です。

1201、サティシュ、25 1202クリシュナ、28 1203 amith、39 1204 javed、23 1205 prudvi、23

16/06/19 15:18:58 ERROR Executor: Exception in task 0.0 in stage 0.0 (TID 0) 
scala.MatchError: 1201 (of class java.lang.Integer) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StringConverter$.toCatalystImpl(CatalystTypeConverters.scala:295) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$StringConverter$.toCatalystImpl(CatalystTypeConverters.scala:294) 
    at org.apache.spark.sql.catalyst.CatalystTypeConverters$CatalystTypeConverter.toCatalyst(CatalystTypeConverters.scala:102) 

Answer 1

スキーマは、すべての列タイプで作成されていますStringTypeとして定義されています。

val schema = StructType(employeeSchemaString.split(",").map(colNmae => StructField(colNmae,StringType,true))) 

ただし、rowRDDにはint型、string型、int型の列があります。ここで

はなぜあなたが分割されている作業コード

val structType= { 
    val id = StructField("id", IntegerType) 
    val name = StructField("name", StringType) 
    val age = StructField("age", IntegerType) 
    new StructType(Array(id, name , age)) 
} 

val rowRdd= employeeRdd.map(row => row.split(",")).map(row => Row(row(0).trim().toInt,row(1),row(2).trim().toInt)) 

sqlContext.createDataFrame(rowRdd,structType). registerTempTable("Employee") 

sqlContext.sql("select * from Employee").show()