2

我是 Spark 和 Scala 的新手。

我需要读取和分析 Spark 中的一个文件，该文件用我的 scala 代码编写，并使用 Kryo 序列化：

import com.esotericsoftware.kryo.Kryo
import com.esotericsoftware.kryo.io.Output

val kryo:Kryo = new Kryo()
val output:Output = new Output(new FileOutputStream("filename.ext",true))

//kryo.writeObject(output, feed) (tested both line)
kryo.writeClassAndObject(output, myScalaObject)

这是用我的对象（myScalaObject）序列化创建文件的伪代码，这是一个复杂的对象。

该文件似乎写得很好，但是当我在 Spark RDD 中读取它时遇到问题

Spark中的伪代码：

val conf = new SparkConf()
    .setMaster("local")
    .setAppName("My application")
    .set("spark.executor.memory", "1g")


conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
conf.set("spark.kryo.registrator", "myScalaObject")

val sc = new SparkContext(conf)

val file=sc.objectFile[myScalaObject]("filename.ext")

val counts = file.count()

当我尝试执行它时，我收到此错误：

org.apache.spark.SparkException: 
Job aborted: Task 0.0:0 failed 1 times (most recent failure: 
Exception failure: java.io.IOException: file: filename.ext not a SequenceFile)

可以在 Spark 中读取这种类型的文件吗？

如果此解决方案不可行，那么创建复杂文件结构以在 Spark 中读取的好解决方案是什么？

谢谢你

4

1 回答 1

3

如果你想用 读取objectFile，用 写出数据saveAsObjectFile。

val myObjects: Seq[MyObject] = ...
val rddToSave = sc.parallelize(myObjects) // Or better yet: construct as RDD from the start.
rddToSave.saveAsObjectFile("/tmp/x")
val rddLoaded = sc.objectFile[MyObject]("/tmp/x")

或者，zsxwing如前所述，您可以创建文件名的 RDD，并用于map读取每个文件名的内容。如果希望将每个文件读入单独的分区，请将文件名并行化到单独的分区中：

def loadFiles(filenames: Seq[String]): RDD[Object] = {
  def load(filename: String): Object = {
    val input = new Input(new FileInputStream(filename))
    return kryo.readClassAndObject(input)
  }
  val partitions = filenames.length
  return sc.parallelize(filenames, partitions).map(load)
}
于 2014-06-02T12:35:28.657 回答