【发布时间】：2018-03-02 11:14:18
【问题描述】：
我的最终 RDD 是这样的
FinalRDD.collect()
Array[(Int, Seq[Iterable[Int]])] = Array((1,List(List(97), List(98), List(99), List(100))), (2,List(List(97, 98), List(97, 99), List(97, 101))), (3,List(List(97, 98, 99),List(99, 102, 103))))
我想将此 RDD 写入以下格式的文本文件
('97'), ('98'), ('100')
('97', '98'), ('97', '99'), List(97, 101)
('97','98', '99'), ('97', '99', '101')
我发现许多网站都建议使用 java.io 中的 PrintWriter 类作为实现此目的的一种选择。这是我尝试过的代码。
val writer = new PrintWriter(new File(outputFName))
def writefunc(chunk : Seq[Iterable[Int]])
{
var n=chunk
print("inside write func")
for(i <- 0 until n.length)
{
writer.print("('"+n(i)+"')"+", ")
}
}
finalRDD.mapValues(list =>writefunc(list)).collect()
我最终得到了如下所示的任务序列化错误
finalRDD.mapValues(list =>writefunc(list)).collect()
org.apache.spark.SparkException: Task not serializable
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:340)
at org.apache.spark.util.ClosureCleaner$.org$apache$spark$util$ClosureCleaner$$clean(ClosureCleaner.scala:330)
at org.apache.spark.util.ClosureCleaner$.clean(ClosureCleaner.scala:156)
at org.apache.spark.SparkContext.clean(SparkContext.scala:2294)
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$mapValues$1.apply(PairRDDFunctions.scala:758)
at org.apache.spark.rdd.PairRDDFunctions$$anonfun$mapValues$1.apply(PairRDDFunctions.scala:757)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:151)
at org.apache.spark.rdd.RDDOperationScope$.withScope(RDDOperationScope.scala:112)
at org.apache.spark.rdd.RDD.withScope(RDD.scala:362)
at org.apache.spark.rdd.PairRDDFunctions.mapValues(PairRDDFunctions.scala:757)
... 50 elided
Caused by: java.io.NotSerializableException: java.io.PrintWriter
Serialization stack:
- object not serializable (class: java.io.PrintWriter, value: java.io.PrintWriter@b0c0abe)
- field (class: $iw, name: writer, type: class java.io.PrintWriter)
- object (class $iw, $iw@31afbb30)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@672ca5ae)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@528ac6dd)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@b772a0e)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@7b11bb43)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@94c2342)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@2bacf377)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@718e1924)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@6d112a64)
- field (class: $iw, name: $iw, type: class $iw)
- object (class $iw, $iw@5747e8e4)
- field (class: $line411.$read, name: $iw, type: class $iw)
- object (class $line411.$read, $line411.$read@59a0616c)
- field (class: $iw, name: $line411$read, type: class $line411.$read)
- object (class $iw, $iw@a375f8f)
- field (class: $iw, name: $outer, type: class $iw)
- object (class $iw, $iw@4e3978ff)
- field (class: $anonfun$1, name: $outer, type: class $iw)
- object (class $anonfun$1, <function1>)
at org.apache.spark.serializer.SerializationDebugger$.improveException(SerializationDebugger.scala:40)
at org.apache.spark.serializer.JavaSerializationStream.writeObject(JavaSerializer.scala:46)
at org.apache.spark.serializer.JavaSerializerInstance.serialize(JavaSerializer.scala:100)
at org.apache.spark.util.ClosureCleaner$.ensureSerializable(ClosureCleaner.scala:337)
... 59 more
我仍处于学习 scala 的阶段。有人可以建议我，如何将“Seq[Iterable[Int]]”对象写入文本文件
【问题讨论】：
-
我认为对于您的用例，Spark 的功能与普通的 Scala 之间存在很大的混淆。对于小型 RDD，您可以通过
finalRDD.collect()将结果收集到内存中，但之后不要在 RDD 上使用任何操作，例如mapValues()或map()。那么你的问题将是 100% 关于 Scala 而不是 Spark -
@IgorK 感谢您的宝贵反馈。我会尽量适应我的代码。
-
如果您的意图是让它适用于潜在的大文件并充分利用 Spark 来保存 RDD - 这是一个完全不同的故事（csv 文件最简单的方法是
finalRDD.saveAsTextFile(...)，但看看如何DataFrames 也被保存了finalRDD.toDF().write.format(...).save(...))。在这种情况下，重新定义序列化格式的细节可能是个坏主意——依赖标准方法。如果您需要文本 - 支持 CSV github.com/databricks/spark-csv -
因此了解您要做什么会很有帮助 - 1) 以任何工作方式将小结果保存在 Scala 代码中或 2) 以可扩展的方式利用 Spark（以便以后可以以适当的数据格式等保存来自多个节点的数据）
-
@IgorK 我会说我正在处理小型数据集，因此想使用 Scala 代码保存最终结果
标签： scala rdd printwriter