本文介绍现在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
}
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)
}
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方法。
全文完 :)
