spark 与flume 1.6.0的示例代码

小编给大家分享一下spark 与flume 1.6.0的示例代码，希望大家阅读完这篇文章之后都有所收获，下面让我们一起去探讨吧！

package hgs.spark.streamingimport org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport org.apache.spark.streaming.StreamingContextimport org.apache.spark.streaming.Secondsimport org.apache.spark.streaming.flume.FlumeUtilsimport org.apache.spark.storage.StorageLevelimport org.apache.spark.HashPartitionerobject SparkStreamingFlumePush {  def main(args: Array[String]): Unit = {    val conf = new SparkConf().setAppName("flume-push").setMaster("local[2]");    val sc = new SparkContext(conf)    val ssc = new StreamingContext(sc,Seconds(5))    ssc.checkpoint("d:\\checkpoint")    val updateFunc=(iter:Iterator[(String,Seq[Int],Option[Int])])=>{    //iter.flatMap(it=>Some(it._2.sum+it._3.getOrElse(0)).map((it._1,_)))//方式一    //iter.flatMap{case(x,y,z)=>{Some(y.sum+z.getOrElse(0)).map((x,_))}}//方式二    iter.flatMap(it=>Some(it._1,(it._2.sum.toInt+it._3.getOrElse(0))))//方式三    }    //总共有两种获取数据的方式，push和poll，这种是push即flume将数据推送给spark 该出的ip、port是spark的ip地址和port    val rds = FlumeUtils.createStream(ssc, "192.168.1.9", 8888, StorageLevel.MEMORY_ONLY)    val result = rds.flatMap(x=>(new String(x.event.getBody.array())).split(" "))    .map(x=>(x,1))    .updateStateByKey(updateFunc, new HashPartitioner(sc.defaultMinPartitions), true)        result.print()        ssc.start()    ssc.awaitTermination()          }}

package hgs.spark.streamingimport org.apache.spark.streaming.StreamingContextimport org.apache.spark.SparkConfimport org.apache.spark.SparkContextimport org.apache.spark.streaming.Secondsimport org.apache.spark.streaming.flume.FlumeUtilsimport java.net.InetAddressimport java.net.InetSocketAddressimport org.apache.spark.storage.StorageLevelimport org.apache.spark.HashPartitioner//spark支持1.6.0的flume版本//同时需要如下三个包 将三个包放到flume的classpath下面 object SparkStreamingFlumePoll {  def main(args: Array[String]): Unit = {   val conf = new SparkConf().setAppName("flume-push").setMaster("local[2]");   val sc = new SparkContext(conf)   val ssc = new StreamingContext(sc,Seconds(5))   ssc.checkpoint("d:\\checkpoint")   val ipSeq =  Seq(new InetSocketAddress("192.168.6.129",8888))   //这种方式通过spark从flume拉取数据   val rds = FlumeUtils.createPollingStream(ssc, ipSeq, StorageLevel.MEMORY_AND_DISK)      val updateFunc=(iter:Iterator[(String,Seq[Int],Option[Int])])=>{    //iter.flatMap(it=>Some(it._2.sum+it._3.getOrElse(0)).map((it._1,_)))//方式一    //iter.flatMap{case(x,y,z)=>{Some(y.sum+z.getOrElse(0)).map((x,_))}}//方式二    iter.flatMap(it=>Some(it._1,(it._2.sum.toInt+it._3.getOrElse(0))))//方式三    }   val result = rds.flatMap(x=>(new String(x.event.getBody.array())).split(" "))    .map(x=>(x,1))    .updateStateByKey(updateFunc, new HashPartitioner(sc.defaultMinPartitions), true)        result.print()        ssc.start()    ssc.awaitTermination()  }}//遇到的错误 scala-library包在flume 的lib下面本来就有，包重复导致的冲突，删除一个

看完了这篇文章，相信你对“spark 与flume 1.6.0的示例代码”有了一定的了解，如果想了解更多相关知识，欢迎关注编程网行业资讯频道，感谢各位的阅读！