首页 > 代码库 > Spark上对SequenceFile的支持

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方法。



全文完 :)