Spark 實現自定義對象sequenceFile方式存儲，讀寫示例（scala編寫）

package com.fuge.bigdata.datahub.analysis

import java.io.{DataInput, DataOutput}

import com.fuge.bigdata.tools.common.utils.SparkUtils
import org.apache.hadoop.io.{NullWritable, WritableComparable}
import org.apache.spark.SparkContext

/**
  * Created by chen xiang on 18-6-13.
  * 一個使用SequenceFile進行存儲讀取的使用示例
  */
object SequenceFileUsage {
  def main(args: Array[String]): Unit = {

    require(args.length == 1)
    // 構建SparkContext對象，封裝過，單獨運行，自行修改后定義
    val sc = new SparkContext(SparkUtils.getSparkConf("SequenceFileUsage"))

    // 獲取路徑參數
    val path = args(0).trim

    // 定義測試數據
    val studentList = List(Student("01", "abc"), Student("02", "baby"), Student("03", "xiang"))

    // 序列化測試數據到RDD,並寫入到bos
    sc.parallelize(studentList)
      .repartition(1)
      // 以NullWritable 為key,構建kv結構.SequenceFile需要kv結構才能存儲,NullWritable不占存儲
      .map(NullWritable.get() -> _)
// 壓縮參數可選用
      .saveAsSequenceFile(s"$path", Option(classOf[GzipCodec]))

    // 讀取剛才寫入的數據
    val studentRdd = sc.sequenceFile(s"$path/part-*", classOf[NullWritable], classOf[Student])
      .map {
        // 讀取數據,並且重新賦值對象
        case (_, y) => Student(y.id, y.name)
      }
      .persist()

    studentRdd
      .foreach(x => println("count: " + x.id + "\t" + x.name))
  }
}

case class Student(var id: String, var name: String) extends WritableComparable[Student] {
  /**
    * 重寫無參構造函數,用於反序列化時的反射操作
    */
  def this() {
    this("", "")
  }

  /**
    * 繼承Comparable接口需要實現的方法,用於比較兩個對象的大小
    */
  override def compareTo(o: Student): Int = {
    var cmp = id compareTo o.id
    if (cmp == 0) {
      cmp = name compareTo o.name
    }
    cmp
  }

  /**
    * 繼承Writable接口需要實現的方法-反序列化讀取結果,並且賦值到對象字段
    * 注意要和write的順序一致
    */
  override def readFields(in: DataInput): Unit = {
    name = in.readUTF()
    id = in.readUTF()
    println("count: " + "\t id = " + id + "\t name = " + name)
  }

  /**
    * 繼承Writable接口需要實現的方法-序列化寫操作,將對象字段值寫入序列化
    * 注意要和readFields的順序一致
    */
  override def write(out: DataOutput): Unit = {
    out.writeUTF(id)
    out.writeUTF(name)
  }
}

 

補充：

1. 自定義的類需要進行序列化，必須都要實現Writable，一般情況下采用實現WritableComparable的方式，並且實現comparaTo，readFields, write方法，並且提供一個無參構造函數

2. readFields和write方法，里面字段的順序要保持一致

3. 遇到集合類型，序列化時需要先將集合長度寫進去，然后再挨個寫集合數據

4. 遇到集合類型，反序列化時需要先讀取集合的長度，然后接收數據，如果集合數據類型是自定義類型，還需要先實例化一個無參構造，然后賦值。

5. SequenceFile需要使用KV結構才能調用存儲，可以使用一個NullWritable來占位，上訴例子中的K值就是使用的NullWritable進行的

6. sequenceFile序列化后占用的存儲空間比較大，有需要的話，可以在存儲的時候加上壓縮算法，具體使用方式可以見上訴的例子