如何编写数据集编码器以支持将函数映射到组织。阿帕奇。火花sql。Scala Spark中的数据集[字符串]

Question

从Spark 1.6迁移到Spark 2.2*会在尝试对查询拼花地板表返回的数据集应用html" target="_blank">方法时出现错误“错误：无法为“数据集”中存储的类型找到编码器。基本类型（Int、String等）。为了证明同样的错误，我对代码进行了过度简化。代码查询拼花地板文件以返回以下数据类型：“org”。阿帕奇。火花sql。Dataset[org.apache.spark.sql.Row]“我应用一个函数来提取字符串和整数，并返回一个字符串。返回以下数据类型：Array[String]接下来，我需要执行大量操作，需要一个单独的函数。在这个测试函数中，我尝试附加一个字符串，产生与我的详细示例相同的错误。我尝试了一些编码器示例和“case”的使用，但没有找到可行的解决方案。如有任何建议/示例，将不胜感激

scala> var d1 = hive.executeQuery(st)
d1: org.apache.spark.sql.Dataset[org.apache.spark.sql.Row] = [cvdt35_message_id_d: string, 
cvdt35_input_timestamp_s: decimal(16,5) ... 2 more fields]

val parseCVDP_parquet = (s:org.apache.spark.sql.Row) => s.getString(2).split("0x" 
(1)+","+s.getDecimal(1);

scala> var d2 =  d1.map(parseCVDP_parquet)
d2: org.apache.spark.sql.Dataset[String] = [value: string]

scala> d2.take(1)
20/03/25 19:01:08 WARN TaskSetManager: Stage 3 contains a task of very large size (131 KB). The 
maximum recommended task size is 100 KB.
res10: Array[String] = Array(ab04006000504304,1522194407.95162)

scala> def dd(s:String){
 | s + "some string"
 | }
dd: (s: String)Unit

scala> var d3 = d2.map{s=> dd(s) }
<console>:47: 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.

为了进一步提炼问题，我相信这个场景（尽管我还没有尝试所有可能的解决方案）可以进一步简化为以下代码：

scala> var test = ( 1 to 3).map( _ => "just some words").toDS()
test: org.apache.spark.sql.Dataset[String] = [value: string]

scala> def f(s: String){
 | s + "hi"
 | }
f: (s: String)Unit

scala> var test2 = test.map{ s => f(s) }
<console>:42: 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.
   var test2 = test.map{ s => f(s) }

Answer 1

第一个解决方案不适用于我的初始（生产）数据集，而是产生错误“org.apache.spark.SparkException:Task not serializable”（有趣的是，两者都存储为相同的数据类型（org.apache.spark.sql.Dataset[String]=[value:String]），我认为这是相关的。我为我的测试数据集提供了另一个解决方案，它消除了最初的编码器错误，如图所示，它实际上可以解决我的玩具问题，而不是生产数据集。我有点困惑，为什么我的应用程序在从1.6版到2.3版spark的过程中被搁置一边，因为我多年来都不需要对我的应用程序进行任何特殊调整，并且已经成功地运行了它，以进行最有可能数万亿次的计算。其他探索包括将我的方法包装为可序列化，探索@transient关键字，利用“org.apache.spark.serializer.KryoSerializer”，将我的方法编写为函数，并将所有变量更改为“vals”（见“stack”上的相关文章）。

scala>  import spark.implicits._
import spark.implicits._

scala> var test = ( 1 to 3).map( _ => "just some words").toDS()
test: org.apache.spark.sql.Dataset[String] = [value: string]

scala> def f(s: String): String = {
 |   val r = s + "hi"
 |   return r
 |   }
 f: (s: String)String

 scala> var d2 =  test.map{s => f(s)}(Encoders.STRING)
 d2: org.apache.spark.sql.Dataset[String] = [value: string]

 scala> d2.take(1)
 res0: Array[String] = Array(just some wordshi)

斯卡拉

Answer 2

我至少有一个简单问题的解决方案（见下文）。我会测试更多。。。。

scala> var test = ( 1 to 3).map( _ => "just some words").toDS()
test: org.apache.spark.sql.Dataset[String] = [value: string]

scala> def f(s: String): String = {
 | val r = s + "hi"
 | return r
 | }
f: (s: String)String

scala> var test2 = test.rdd.map{ s => f(s) }
test2: org.apache.spark.rdd.RDD[String] = MapPartitionsRDD[17] at map at <console>:43

scala> test2.take(1)
res9: Array[String] = Array(just some wordshi)