Error: Unable to find encoder for type stored in a Dataset.  Primitive types (Int, String, etc) and Product types (case classes) are supported by importing spark.implicits._  Support for serializing other types will be added in future releases.

錯誤:無法找到儲存在資料集中的型別的編碼器。原始型別(Int,String等)和產品型別(case類)由匯入spark.implicits支援。支援對其他型別的序列化，將在以後的版本中新增。

問題解決如下：

在開發過程中，我想將已經儲存在HDFS上的文字檔案儲存成CSV格式的檔案。

以前檔案格式如下（這裡使用自己造的假資料）每一列所對應的是 id,name,no,sp,ep
3303龍順 JD8赤壁 湛江
5426程凡 G58龍巖 苗栗

我想處理成下面的格式，因為下面的格式是CSV預設格式
id,name,no,sp,ep
1309,項敬,BKZ,韶關,湖北
3507,寧風晨,KY7,河源,資陽

處理程式碼如下

def main(args: Array[String]) {
	val sparkSession = SparkSession.builder().appName("Spark shell").getOrCreate()
	//檔案路徑
	val path = "hdfs://master:9000/TestData/aviation9"
	//儲存路徑
	val savePath = "hdfs://master:9000/TestData/aviation10/"
	val file = sparkSession.read.textFile(path)
	//處理資料，拆分
	val rd = file.map(line => {
	  val arr = line.split("\t")
	  (arr(0), arr(1), arr(2), arr(3), arr(4))
	})
	//給DataFrame新增表頭，
	val res = rd.toDF("id", "name", "no", "sp", "ep")
	//儲存有表的檔案
	res.repartition(1).write.mode(SaveMode.Append).format("csv").option("header", true).save(savePath)
}
 

準備打包執行，在打包的時候總是出現上面的錯誤

無法找到儲存在資料集中的型別的編碼器。
原始型別(Int,String等)和產品型別(case類)由匯入spark.implicits支援。
支援對其他型別的序列化，將在以後的版本中新增。
所以需要我們自己在Dataset中新增元組這樣的編碼

解決辦法：處理資料之間新增下面一行程式碼
implicit val matchError = org.apache.spark.sql.Encoders.tuple( Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING)

最後的全部程式碼是

def main(args: Array[String]) {
	val sparkSession = SparkSession.builder().appName("Spark shell").getOrCreate()
	//檔案路徑
	val path = "hdfs://master:9000/TestData/aviation9"
	//儲存路徑
	val savePath = "hdfs://master:9000/TestData/aviation10/"
	val file = sparkSession.read.textFile(path)
	implicit val matchError = org.apache.spark.sql.Encoders.tuple( Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING)
	//處理資料，拆分
	val rd = file.map(line => {
	  val arr = line.split("\t")
	  (arr(0), arr(1), arr(2), arr(3), arr(4))
	})
	//給DataFrame新增表頭，
	val res = rd.toDF("id", "name", "no", "sp", "ep")
	//儲存有表的檔案
	res.repartition(1).write.mode(SaveMode.Append).format("csv").option("header", true).save(savePath)
}

在處理這個錯誤中，發現編碼器中預設的tuple最多是5個元素，那如果我們有很多列資料該怎麼處理呢？
比如現在的資料是下面這樣的，多了一列time
5426程凡 G562013-12-24 17:26:23龍巖苗栗
4413相承 TV72014-04-09 20:44:25北票開原

如果在用下面的程式碼又出現同樣的錯誤
implicit val matchError = org.apache.spark.sql.Encoders.tuple( Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING, Encoders.STRING)

因為在Encoders中最多支援5個元素的tuple，我們需要將DataSet轉成處理成Row型別的，最後轉成RDD用資料和表頭建立一個DataFrame

def main(args: Array[String]) {
    val sparkSession = SparkSession.builder().appName("Spark shell").getOrCreate()
    val fields = "id,name,no,time,sp,ep"
    val path = "hdfs://master:9000/TestData/aviation9"
    val savePath = "hdfs://master:9000/TestData/aviation10/"
    
    val file: Dataset[String] = sparkSession.read.textFile(path)
    
    implicit val matchError = org.apache.spark.sql.Encoders.kryo[Row]
    //處理成Row
    val ds = file.map(x => {
      val arr = x.split("\t")
      Row.fromSeq(arr.toSeq)
    })
    //建立表頭
    val field_array = fields.split(",")
    val schema = StructType(field_array.map(fieldName => StructField(fieldName, StringType, true)))
    //建立DataFrame
    val df = sparkSession.createDataFrame(ds.rdd, schema)
    df.repartition(1).write.mode(SaveMode.Append).format("csv").option("header", true).save(savePath)
}