DataSet資料集在使用sql()時，無法使用map，flatMap等轉換運算元的解決辦法
摘要
我們在使用spark的一個流程是：利用spark.sql()函式把資料讀入到記憶體形成DataSet[Row]（DataFrame）由於Row是新的spark資料集中無法實現自動的編碼，需要對這個資料集進行編碼，才能利用這些運算元進行相關的操作，如何編碼是一個問題，在這裡就把這幾個問題進行總結一下。報的錯誤：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.
報這個錯誤一般就是我們在使用運算元時其返回值的資料型別往往不是spark通過自身的反射能完成的自動編碼部分，比如通過map運算元，我們在map運算元的函式的返回值型別是Map型別的，就會出現上面的問題，因為Map集合類不在：基本的型別和String，case class和元組的範圍之內，spark內部不能通過反射完成自動編碼。
出現這個問題的原因
spark2.0以後的版本採用的是新的分散式資料集DataSet，其中DataFrame是DataSet[Row]的別名形式。而新的資料集採用了很多的優化，其中一個就是利用了Tungsten execution engine的計算引擎，這個計算引擎採用了很多的優化。其中一個就是自己維護了一個記憶體管理器，從而使計算從java jvm解脫出來了，使得記憶體的優化得到了很大的提升。同時新的計算引擎，把資料儲存在記憶體中是以二進位制的形式儲存的，大部分所有的計算都是在二進位制資料流上進行的，不需要把二進位制資料流反序列化成java物件，然後再把計算的結果序列化成二進位制資料流，而是直接在二進位制流上進行操作，這樣的情況就需要我們存在一種機制就是java物件到二進位制資料流的對映關係，不然我們不知道二進位制流對應的資料物件是幾個位元組，spark這個過程是通過Encoders來完成的，spark自身通過反射完成了一部分的自動編碼過程：基本的型別和String，case class和元組，對於其他的集合型別或者我們自定義的類，他是無法完成這樣的編碼的。需要我們自己定義這樣的編碼也就是讓其擁有一個schema。
解決這個問題方式
方法一：
這樣就是把其轉化為RDD，利用RDD進行操作，但是不建議用這個，相對於RDD，DataSet進行了很多的底層優化，擁有很不錯效能
val orderInfo1 = spark.sql(
"""
|SELECT
|o.id,
|o.user_id
|FROM default.api_order o
|limit 100
""".stripMargin).rdd.map(myfunction)方法二：
讓其自動把DataSet[Row]轉化為DataSet[P]，如果Row裡面有複雜的型別出現的話。
case class Orders(id: String, user_id: String)
//這個case class要定義在我們的單例物件的外面
object a {
def main(args: Array[String]): Unit ={
import spark.implicits._
val orderInfo1 = spark.sql(
"""
|SELECT
|o.id,
|o.user_id
|FROM default.api_order o
|limit 100
""".stripMargin).as[Orders].map(myfunction)
}
}方式三：
自定義一個schema，然後利用RowEncoder進行編碼。這只是一個例子，裡面的型別其實都可以通過spark的反射自動完成編碼過程。
import spark.implicits._
val schema = StructType(StructType(Seq(StructField("id",StringType,true),StructField("user_id",StringType,true))))
val encoders = RowEncoder(schema)
val orderInfo1 = spark.sql(
"""
|SELECT
|o.id,
|o.user_id
|FROM default.api_order o
|limit 100
""".stripMargin).map(row => row)(encoders)方法四：
直接利用scala的模式匹配的策略case Row來進行是可以通過的，原因是case Row()scala模式匹配的知識，這樣可以知道集合Row裡面擁有多少個基本的型別，則可以通過scala就可以完成對Row的自動編碼，然後可以進行相應的處理。
import spark.implicits._
val orderInfo1 = spark.sql(
"""
|SELECT
|o.id,
|o.user_id
|FROM default.api_order o
|limit 100
""".stripMargin).map{case Row(id: String, user_id: String) => (id,user_id)}
這個得到的schema為：
orderInfo1: org.apache.spark.sql.Dataset[(String, String)] = [_1: string, _2: string]
如果換成這樣：
val orderInfo1 = spark.sql(
"""
|SELECT
|o.id,
|o.user_id
|FROM default.api_order o
|limit 100
""".stripMargin).map{case Row(id: String, user_id: String) => List(id,user_id)}
得到的schema為：
orderInfo1: org.apache.spark.sql.Dataset[List[String]] = [value: array<string>]
可以看出：spark是把元祖看成case class一種特殊形式擁有，schame的欄位名稱為_1,_2這樣的特殊case clase