Spark 1.6 中的数据集答案

【问题标题】：Dataset in Spark 1.6Spark 1.6 中的数据集
【发布时间】：2016-04-15 10:21:57
【问题描述】：

我正在评估将现有的 RDD 代码替换为数据集。对于我的一个用例，我无法将数据集映射到另一个案例类。

这就是我想要做的......

case class MyMap(map: Map[String, String])

case class V1(a: String, b: String){
  def toMyMap: MyMap = {
    MyMap(Map(a->b))
  }

  def toStr: String = {
    a
  }
}

object MyApp extends App {
//Get handle to sqlContext and other useful stuff here.
val df1 = sqlContext.createDataset(Seq(V1("2015-05-01", "data1"), V1("2015-05-01", "data2"))).toDF()
df1.as[V1].map(_.toMyMap).show() //Errors out. Added the exception below.
df1.as[V1].map(_.toStr).show() //Works fine.
}

任何帮助将不胜感激。

以下例外：

线程“主”org.apache.spark.SparkException 中的异常：作业由于阶段失败而中止：任务不可序列化： java.io.NotSerializableException： scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$1 序列化栈： - 对象不可序列化（类：scala.reflect.runtime.SynchronizedSymbols$SynchronizedSymbol$$anon$1，值：包语言） - 字段（类：scala.reflect.internal.Types$ThisType，名称：sym，类型：class scala.reflect.internal.Symbols$Symbol） - 对象（类 scala.reflect.internal.Types$UniqueThisType，java.lang.type） - 字段（类：scala.reflect.internal.Types$TypeRef，名称：pre，类型：类 scala.reflect.internal.Types$Type） - 对象（类 scala.reflect.internal.Types$ClassNoArgsTypeRef，字符串） - 字段（类：scala.reflect.internal.Types$TypeRef，名称：规范化，类型：类 scala.reflect.internal.Types$Type） - 对象（类 scala.reflect.internal.Types$AliasNoArgsTypeRef，字符串） - 字段（类：org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$6，名称： keyType$1，类型：类 scala.reflect.api.Types$TypeApi) - 对象（类 org.apache.spark.sql.catalyst.ScalaReflection$$anonfun$6, ） - 字段（类：org.apache.spark.sql.catalyst.expressions.MapObjects，名称：函数，类型：接口 scala.Function1） - 对象（类 org.apache.spark.sql.catalyst.expressions.MapObjects, mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType)) - 字段（类：org.apache.spark.sql.catalyst.expressions.Invoke，名称：targetObject，类型：类 org.apache.spark.sql.catalyst.expressions.Expression） - 对象（类 org.apache.spark.sql.catalyst.expressions.Invoke，invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true)),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;))) - writeObject 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.List$SerializationProxy，scala.collection.immutable.List$SerializationProxy@7e78c3cf） - writeReplace 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.$colon$colon, List(invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)), 调用(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),valueArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)))) - 字段（类：org.apache.spark.sql.catalyst.expressions.StaticInvoke，名称：参数，类型：接口 scala.collection.Seq) - 对象（类 org.apache.spark.sql.catalyst.expressions.StaticInvoke，静态调用（类 org.apache.spark.sql.catalyst.util.ArrayBasedMapData$,ObjectType(interface scala.collection.Map),toScalaMap,invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),valueArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),true)) - writeObject 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.List$SerializationProxy，scala.collection.immutable.List$SerializationProxy@377795c5） - writeReplace 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.$colon$colon, List(staticinvoke(class org.apache.spark.sql.catalyst.util.ArrayBasedMapData$,ObjectType(interface scala.collection.Map),toScalaMap,invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),valueArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),true))) - 字段（类：org.apache.spark.sql.catalyst.expressions.NewInstance，名称：参数，类型：接口 scala.collection.Seq) - 对象（类 org.apache.spark.sql.catalyst.expressions.NewInstance，新实例（类收集器。MyMap，staticinvoke（类 org.apache.spark.sql.catalyst.util.ArrayBasedMapData$,ObjectType(interface scala.collection.Map),toScalaMap,invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),keyArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),invoke(mapobjects(,invoke(upcast('map,MapType(StringType,StringType,true),- 字段（类：“scala.collection.immutable.Map”，名称：“map”），- 根班级： "collector.MyMap"),valueArray,ArrayType(StringType,true)),StringType),array,ObjectType(class [Ljava.lang.Object;)),true),false,ObjectType(类收集器.MyMap），无）） - 字段（类：org.apache.spark.sql.catalyst.encoders.ExpressionEncoder，名称： fromRowExpression，类型：类 org.apache.spark.sql.catalyst.expressions.Expression） - 对象（类 org.apache.spark.sql.catalyst.encoders.ExpressionEncoder，类[地图#ExprId（9,255a02aa-f2fa-482d-8cd1-63e2d4d08b30）：地图]） - 字段（类：org.apache.spark.sql.execution.MapPartitions，名称：uEncoder，类型：类 org.apache.spark.sql.catalyst.encoders.ExpressionEncoder) - 对象（类 org.apache.spark.sql.execution.MapPartitions，!MapPartitions，类[a[0]：字符串，b[0]：字符串]，类[地图#ExprId（9,255a02aa-f2fa-482d-8cd1-63e2d4d08b30）：地图], [地图#13] +- LocalTableScan [a#2,b#3], [[0,180000000a,2800000005,2d35302d35313032,3130,3161746164],[0,180000000a,2800000005,2d35302d35313032,3130,3267 ) - 字段（类：org.apache.spark.sql.execution.MapPartitions$$anonfun$8，名称：$outer，类型：类 org.apache.spark.sql.execution.MapPartitions) - 对象（类 org.apache.spark.sql.execution.MapPartitions$$anonfun$8, ） - 字段（类：org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1，名称：f$22，类型：接口 scala.Function1) - 对象（类 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1， ) - 字段（类：org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$21，名称：$outer，类型：类 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1) - 对象（类 org.apache.spark.rdd.RDD$$anonfun$mapPartitionsInternal$1$$anonfun$apply$21， ) - 字段（类：org.apache.spark.rdd.MapPartitionsRDD，名称：f，类型：接口 scala.Function3） - 对象（在 CollectorSparkTest.scala:50 上展示的类 org.apache.spark.rdd.MapPartitionsRDD、MapPartitionsRDD[1]） - 字段（类：org.apache.spark.NarrowDependency，名称：rdd，类型：类 org.apache.spark.rdd.RDD） - 对象（类 org.apache.spark.OneToOneDependency，org.apache.spark.OneToOneDependency@110f15b7） - writeObject 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.List$SerializationProxy，scala.collection.immutable.List$SerializationProxy@6bb23696） - writeReplace 数据（类：scala.collection.immutable.List$SerializationProxy） - 对象（类 scala.collection.immutable.$colon$colon，列表（org.apache.spark.OneToOneDependency@110f15b7）） - 字段（类：org.apache.spark.rdd.RDD，名称：org$apache$spark$rdd$RDD$$dependencies，类型：接口 scala.collection.seq) - 对象（在 CollectorSparkTest.scala:50 上展示的类 org.apache.spark.rdd.MapPartitionsRDD、MapPartitionsRDD[2]） - 字段（类：scala.Tuple2，名称：_1，类型：类 java.lang.Object） - 对象（class scala.Tuple2, (MapPartitionsRDD[2] at show at CollectorSparkTest.scala:50,)) at org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1431) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1419) 在 org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1418) 在 scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) 在 scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.abortStage（DAGScheduler.scala:1418）在 org.apache.spark.scheduler.DAGScheduler.submitMissingTasks(DAGScheduler.scala:1010) 在 org.apache.spark.scheduler.DAGScheduler.org$apache$spark$scheduler$DAGScheduler$$submitStage(DAGScheduler.scala:921) 在 org.apache.spark.scheduler.DAGScheduler.handleJobSubmitted(DAGScheduler.scala:861) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.doOnReceive(DAGScheduler.scala:1607) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1599) 在 org.apache.spark.scheduler.DAGSchedulerEventProcessLoop.onReceive(DAGScheduler.scala:1588) 在 org.apache.spark.util.EventLoop$$anon$1.run(EventLoop.scala:48) 在 org.apache.spark.scheduler.DAGScheduler.runJob(DAGScheduler.scala:620) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1832) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1845) 在 org.apache.spark.SparkContext.runJob(SparkContext.scala:1858) 在 org.apache.spark.sql.execution.SparkPlan.executeTake(SparkPlan.scala:212) 在 org.apache.spark.sql.execution.Limit.executeCollect(basicOperators.scala:165) 在 org.apache.spark.sql.execution.SparkPlan.executeCollectPublic(SparkPlan.scala:174) 在 org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1538) 在 org.apache.spark.sql.DataFrame$$anonfun$org$apache$spark$sql$DataFrame$$execute$1$1.apply(DataFrame.scala:1538) 在 org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56) 在 org.apache.spark.sql.DataFrame.withNewExecutionId(DataFrame.scala:2125) 在 org.apache.spark.sql.DataFrame.org$apache$spark$sql$DataFrame$$execute$1(DataFrame.scala:1537) 在 org.apache.spark.sql.DataFrame.org$apache$spark$sql$DataFrame$$collect(DataFrame.scala:1544) 在 org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1414) 在 org.apache.spark.sql.DataFrame$$anonfun$head$1.apply(DataFrame.scala:1413) 在 org.apache.spark.sql.DataFrame.withCallback(DataFrame.scala:2138) 在 org.apache.spark.sql.DataFrame.head(DataFrame.scala:1413) 在 org.apache.spark.sql.DataFrame.take(DataFrame.scala:1495) 在 org.apache.spark.sql.DataFrame.showString(DataFrame.scala:171) 在 org.apache.spark.sql.DataFrame.show(DataFrame.scala:394) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:228) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:192) 在 org.apache.spark.sql.Dataset.show(Dataset.scala:200)

【问题讨论】：

标签： apache-spark apache-spark-sql spark-dataframe apache-spark-dataset

【解决方案1】：

问题在于 scala Map 类不可序列化，因此 Dataset API 无法自动生成合适的编码器。我建议将映射转换为字符串，然后解析字符串并转换回映射（假设您将字符串存储在映射中）。

Dataset API 也可能不是最佳选择。我写了this article，这可能很有趣。

【讨论】：

【解决方案2】：

我认为您实际上可能会点击SPARK-12696，它已在 spark/master 中修复。我希望在不久的将来发布 1.6.1，其中应该包含这个补丁。

【讨论】：