Spark中将对象序列化存储到hdfs

废话不多说, 直接贴代码了. spark1.4 + hbase0.98

1import org.apache.spark.storage.StorageLevel
2import scala.collection.JavaConverters._
3import java.io.File
4import java.io.FileInputStream
5import java.io.FileOutputStream
6import java.io.ObjectInputStream
7import java.io.ObjectOutputStream
8import java.net.URI
9import java.util.Date
10import org.ansj.library.UserDefineLibrary
11import org.ansj.splitWord.analysis.NlpAnalysis
12import org.ansj.splitWord.analysis.ToAnalysis
13import org.apache.hadoop.fs.FSDataInputStream
14import org.apache.hadoop.fs.FSDataOutputStream
15import org.apache.hadoop.fs.FileSystem
16import org.apache.hadoop.fs.FileUtil
17import org.apache.hadoop.fs.Path
18import org.apache.hadoop.hbase.client._
19import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor, TableName}
20import org.apache.hadoop.hbase.filter.FilterList
21import org.apache.hadoop.hbase.filter.PageFilter
22import org.apache.hadoop.hbase.filter.RegexStringComparator
23import org.apache.hadoop.hbase.filter.SingleColumnValueFilter
24import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp
25import org.apache.hadoop.hbase.mapreduce.TableInputFormat
26import org.apache.hadoop.hbase.protobuf.ProtobufUtil
27import org.apache.hadoop.hbase.util.{Base64, Bytes}
28import com.feheadline.fespark.db.Neo4jManager
29import com.feheadline.fespark.util.Env
30import org.apache.spark.SparkConf
31import org.apache.spark.SparkContext
32import org.apache.spark.rdd._
33import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}
34import scala.math.log
35import scala.io.Source
36
37object Word2VecDemo {
38
39  def convertScanToString(scan: Scan) = {
40    val proto = ProtobufUtil.toScan(scan)
41    Base64.encodeBytes(proto.toByteArray)
42  }
43
44  def main(args: Array[String]): Unit = {
45    val sparkConf = new SparkConf().setAppName("Word2Vec Demo")
46    sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
47    sparkConf.set("spark.kryoserializer.buffer", "256m")
48    sparkConf.set("spark.kryoserializer.buffer.max","2046m")
49    sparkConf.set("spark.akka.frameSize", "500")
50    sparkConf.set("spark.rpc.askTimeout", "30")
51    
52
53    val sc = new SparkContext(sparkConf)
54    val hbaseConf = HBaseConfiguration.create()
55    hbaseConf.set("hbase.zookeeper.quorum", "myzookeeper")
56
57    hbaseConf.set(TableInputFormat.INPUT_TABLE, "crawled")
58
59    val scan = new Scan()
60    val filterList:FilterList = new FilterList(FilterList.Operator.MUST_PASS_ALL)
61    
62    val comp:RegexStringComparator = new RegexStringComparator(""".{1500,}""")
63    
64    val articleFilter:SingleColumnValueFilter = new SingleColumnValueFilter(
65    "data".getBytes,
66    "article".getBytes,
67    CompareOp.EQUAL,
68    comp
69    )
70    
71    filterList.addFilter(articleFilter)
72    filterList.addFilter(new PageFilter(100))
73    
74    scan.setFilter(filterList)
75    scan.setCaching(50)
76    scan.setCacheBlocks(false)
77    hbaseConf.set(TableInputFormat.SCAN,convertScanToString(scan))
78
79    val crawledRDD = sc.newAPIHadoopRDD(
80      hbaseConf,
81      classOf[TableInputFormat],
82      classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable],
83      classOf[org.apache.hadoop.hbase.client.Result]
84    )
85 
86    val articlesRDD = crawledRDD.filter{
87      case (_,result) => {
88          val content = Bytes.toString(result.getValue("data".getBytes,"article".getBytes))
89          content != null
90      }
91    }
92
93    val wordsInDoc = articlesRDD.map{
94      case (_,result) => {
95          val content = Bytes.toString(result.getValue("data".getBytes,"article".getBytes))
96          if(content!=null)ToAnalysis.parse(content).asScala.map(_.getName).toSeq
97          else Seq("")
98      }
99    }
100    
101    val fitleredWordsInDoc = wordsInDoc.filter(_.nonEmpty)
102    
103    val word2vec = new Word2Vec()
104    val model = word2vec.fit(fitleredWordsInDoc)
105   
106    //---------------------------------------重点看这里-------------------------------------------------------------
107    //将上面的模型存储到hdfs
108    val hadoopConf = sc.hadoopConfiguration
109    hadoopConf.set("fs.defaultFS", "hdfs://myhadoop:9000/")
110    val fileSystem = FileSystem.get(hadoopConf)
111    val path = new Path("/user/hadoop/data/mllib/word2vec-object")
112    val oos = new ObjectOutputStream(new FSDataOutputStream(fileSystem.create(path)))
113    oos.writeObject(model)
114    oos.close
115    
116    //这里示例另外一个程序直接从hdfs读取序列化对象使用模型
117    val ois = new ObjectInputStream(new FSDataInputStream(fileSystem.open(path)))
118    val sample_model = ois.readObject.asInstanceOf[Word2VecModel]
119   
120    /*
121    * //你还可以将序列化文件从hdfs放到本地, scala程序使用模型
122    * import java.io._
123    * import org.apache.spark.mllib.feature.{Word2Vec, Word2VecModel}
124    * val ois = new ObjectInputStream(new FileInputStream("/home/cherokee/tmp/word2vec-object"))
125    * val sample_model = ois.readObject.asInstanceOf[Word2VecModel]
126    * ois.close
127    */
128    //--------------------------------------------------------------------------------------------------------------
129  }
130}