Spark上對SequenceFile的支持

本文介紹現在Spark提供的API裏對Hadoop SequenceFile的讀寫支持，涉及到的類和使用方式，包括scala環境和python環境。

Scala環境下的支持

spark下涉及到seqeucenfile的讀寫，主要有兩類體係，第一類是帶'sequenceFIle'的方法，第二類是帶'ObjectFile'的方法。

以下是SparkContext下的三個讀取seqeucenfile的方法，除了指定path路徑外，還需要聲明key，value對應的hadoop writable類，此外還可以指定分區數。

  def sequenceFile[K, V](path: String,
      keyClass: Class[K],
      valueClass: Class[V],
      minPartitions: Int
      ): RDD[(K, V)] = {
    val inputFormatClass = classOf[SequenceFileInputFormat[K, V]]
    hadoopFile(path, inputFormatClass, keyClass, valueClass, minPartitions)
  }

  def sequenceFile[K, V](path: String, keyClass: Class[K], valueClass: Class[V]
      ): RDD[(K, V)] =
    sequenceFile(path, keyClass, valueClass, defaultMinPartitions)

  def sequenceFile[K, V]
       (path: String, minPartitions: Int = defaultMinPartitions)
       (implicit km: ClassTag[K], vm: ClassTag[V],
        kcf: () => WritableConverter[K], vcf: () => WritableConverter[V])
      : RDD[(K, V)] = {
    val kc = kcf()
    val vc = vcf()
    val format = classOf[SequenceFileInputFormat[Writable, Writable]]
    val writables = hadoopFile(path, format,
        kc.writableClass(km).asInstanceOf[Class[Writable]],
        vc.writableClass(vm).asInstanceOf[Class[Writable]], minPartitions)
    writables.map { case (k, v) => (kc.convert(k), vc.convert(v)) }
  }

讀取的時候的K，V，可以直接寫org.apache.hadoop.io.BytesWritable這樣的類，也可以寫基本類型，如Int，String，會被隱式轉換成對應的org.apache.hadoop.io.IntWritable，org.apache.hadoop.io.Text。

另一方麵，第二類方法是objectFile方法

  def objectFile[T: ClassTag](
      path: String,
      minPartitions: Int = defaultMinPartitions
      ): RDD[T] = {
    sequenceFile(path, classOf[NullWritable], classOf[BytesWritable], minPartitions)
      .flatMap(x => Utils.deserialize[Array[T]](x._2.getBytes, Utils.getContextOrSparkClassLoader))
  }

該方法對應的是RDD裏麵saveAsObjectFile的方法，key class是NullWritable，value class是BytesWritable，且反序列化過程也指明好了，利用的是Utils裏的序列化方法，可以看到，裏麵的序列化利用的是java原生的序列化方式，如下：

  /** Deserialize an object using Java serialization */
  def deserialize[T](bytes: Array[Byte]): T = {
    val bis = new ByteArrayInputStream(bytes)
    val ois = new ObjectInputStream(bis)
    ois.readObject.asInstanceOf[T]
  }

  /** Deserialize an object using Java serialization and the given ClassLoader */
  def deserialize[T](bytes: Array[Byte], loader: ClassLoader): T = {
    val bis = new ByteArrayInputStream(bytes)
    val ois = new ObjectInputStream(bis) {
      override def resolveClass(desc: ObjectStreamClass) =
        Class.forName(desc.getName, false, loader)
    }
    ois.readObject.asInstanceOf[T]
  }

下麵先繼續介紹sequencefile的寫方法，調用的是RDD的saveAsObjectFile方法，如下，

  /**
   * Save this RDD as a SequenceFile of serialized objects.
   */
  def saveAsObjectFile(path: String) {
    this.mapPartitions(iter => iter.grouped(10).map(_.toArray))
      .map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x))))
      .saveAsSequenceFile(path)
  }

對應到SparkContext裏的objectFile方法，RDD的save也指定了key、value的writable類，利用的是同一套序列化方式，

  /** Serialize an object using Java serialization */
  def serialize[T](o: T): Array[Byte] = {
    val bos = new ByteArrayOutputStream()
    val oos = new ObjectOutputStream(bos)
    oos.writeObject(o)
    oos.close()
    bos.toByteArray
  }

回過頭繼續看RDD的saveAsObjectFile方法裏，在做完map操作後，其實是隱式生成了SequenceFileRDDFunction類，具體implicit的定義在SparkContext裏：

  implicit def rddToSequenceFileRDDFunctions[K <% Writable: ClassTag, V <% Writable: ClassTag](
      rdd: RDD[(K, V)]) =
    new SequenceFileRDDFunctions(rdd)

所以其實調用的是SequenceFileRDDFunction的saveAsSequenceFile方法，在該方法裏，最終調用的是RDD的saveHadoopFile這個老的hadoop file方法，並且傳遞了SequenceFileOutputFormat這個format給saveHadoopFile方法，從而完成hadoop file的寫入。

下麵是一個簡單的讀寫sequencefile的例子，可以自己在spark-shell裏嚐試下

val list = List("ss", "rdd", "egerg", 324, 123)
val r = sc.makeRDD(list, 1)
r.saveAsObjectFile("hdfs:/your/path/list")

val file = sc.sequenceFile[Null,org.apache.hadoop.io.BytesWritable]("hdfs:/your/path/list/part-00000")
val bw = file.take(1).apply(0)._2
val bs = bw.getBytes

import java.io._
val bis = new ByteArrayInputStream(bs)
val ois = new ObjectInputStream(bis)
ois.readObject

上麵在讀出來反序列化的時候，我模仿Utils裏的方式利用java.io手動反序列出來了。

其實也可以模仿RDD的那個saveAsObjectFile方法，自己設定key，value，序列化方式等設置，改造下麵這段代碼裏的transformation過程，

  def saveAsObjectFile(path: String) {
    this.mapPartitions(iter => iter.grouped(10).map(_.toArray))
      .map(x => (NullWritable.get(), new BytesWritable(Utils.serialize(x))))
      .saveAsSequenceFile(path)
  }

如上所說，已經比較清晰地說明了sequencfile讀寫的來龍去脈了，也給出了簡單的讀寫例子，包括如何聲明writable類型，甚至可以模仿RDD的saveAsObjectFile方法做到更好的讀寫控製。

pyspark下的支持

python環境下的支持可以參考這個PR，目前已經合進社區的master分支裏了。之前python環境下隻支持textFile。這個PR除了支持sequenceFile的讀寫外，還支持了hadoop下其他format文件的讀取。主要是增加了PythonRDD裏的sequenceFile、newAPIHadoopFile等的支持，然後在python/pyspark/context.py裏增加了上下文裏的相應方法支持，使得pyspark裏也可以得到豐富的hadoop file讀取的支持。

使用的話，直接讀取就可以了

lines = sc.sequenceFile("hdfs:/your/path/list/part-00000")

總結

本文介紹了spark對hadoop sequencefile的讀寫支持，實現方式以及簡單的使用方法。sequencefile和textfile類似，在上下文裏有直接提供讀取方法，但最終走的還是hadoopFile方法。

全文完 :)

最後更新：2017-04-03 05:39:04

Spark上對SequenceFile的支持

Scala環境下的支持

pyspark下的支持

總結

上一篇： Linux LVM硬盤管理及LVM擴容

下一篇： C++ 模板中的template typename 和template class的區別

相關內容

熱門內容

最新內容