flume整合sparkstreaming

zhouyi

这里写一个flume整合sparkstreaming的例子
我这里使用scala ide和maven的方式
spark用的1.2版本，scala是2.10版本，flume使用最新版的1.6
整合的第一步，要加上flume的maven依赖  <dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming-flume_2.10</artifactId>
<version>1.2.1</version>
</dependency>
在这里，不进行spark环境的搭建，为了方便起见，直接使用scala ide本地local模式运行
代码就已单词计数为例package cn.han
import org.apache.spark.SparkConf
import org.apache.spark.streaming.Seconds
import org.apache.spark.streaming.StreamingContext
import org.apache.spark.streaming.StreamingContext._
import org.apache.spark.streaming.flume._
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.storage.StorageLevel
object MySpark {
def main(args: Array[String]): Unit = {
val sc = new SparkContext("local[2]", "Spark Streaming Flume Integration")  
//创建StreamingContext，3秒一个批次  
val ssc = new StreamingContext(sc, Seconds(3))  
val hostname = "192.168.5.102"
val port = 11111  
val flumeStream = FlumeUtils.createStream(ssc, hostname, port)  
val rss2=flumeStream.flatMap(x =>{
val by=x.event.getBody.array()
val str=new String(by)
val sb=str.split(" ")
sb
})
val rdd3=rss2.map(x=>(x,1))
val rdd4=rdd3.reduceByKey(_+_)
rdd4.print()
//开始运行  
ssc.start()  
//计算完毕退出  
ssc.awaitTermination()  
sc.stop()  
}
}
这里有一个需要及其注意的地方，就是local的参数，必须要大于1，除了receive也需要占用一个，如果写一个，就接收不到数据
然后，是flume的一些配置，首先，下载最新版的flume，解压，直接在conf目录下新建
agent3

直接启动flumebin/flume-ng agent --conf ./conf/ -f conf/agent3 -Dflume.root.logger=DEBUG,console -n agent3

flume这边就启动完成了
然后还需要一个flume的源就是source,这里配置的是avro，就使用log4j的方式直接写入flume
首先引入log4j放入maven，这里面需要下载一个jar包引入才行flume-ng-log4jappender-1.5.0-cdh5.1.3-jar-with-dependencies.jar
然后需要修改下log4j.properties文件log4j.rootLogger=INFO,flume
log4j.appender.flume = org.apache.flume.clients.log4jappender.Log4jAppender
log4j.appender.flume.Hostname = 192.168.5.220
log4j.appender.flume.Port = 41414
log4j.appender.flume.UnsafeMode = true
这时候在写个简单的测试类进行测试package cn.han;
import org.apache.log4j.Logger;
public class MyLogh {
private static Logger logger = Logger.getLogger(MyLogh.class);
public static void main(String[] args) throws Exception{
while(true){
for(int i=0;i<120;i++){
logger.info("hello you");
}
Thread.sleep(1000);
}
}
}
这样运行spark个程序就可以看到结果，每3秒执行一次job