转发自玖疯的博客
http://www.cnblogs.com/lxf20061900/p/3866252.html
Spark Streaming是一个新的实时计算的利器,而且还在快速的发展。它将输入流切分成一个个的DStream转换为RDD,从而可以使用Spark来处理。它直接支持多种数据源:Kafka, Flume, Twitter, ZeroMQ , TCP sockets等,有一些可以操作的函数:map
, reduce
, join
, window等。
本文将Spark Streaming和Flume-NG进行对接,然后以官方内置的JavaFlumeEventCount作参考,稍作修改然后放到集群上去运行。
一、下载spark streaming的flume插件包,我们这里的spark版本是1.0.0(standlone),这个插件包的版本选择spark-streaming-flume_2.10-1.0.1.jar,这个版本修复了一个重要的bug,参考下面参考中的7。
二、把spark的编译后的jar包以及上面flume的插件,放入工程,编写如下类(参考8中的例子修改而来),代码如下:
1 package com.spark_streaming; 2 3 import org.apache.spark.SparkConf; 4 import org.apache.spark.api.java.function.Function; 5 import org.apache.spark.streaming.*; 6 import org.apache.spark.streaming.api.java.*; 7 import org.apache.spark.streaming.flume.FlumeUtils; 8 import org.apache.spark.streaming.flume.SparkFlumeEvent; 9 10 public final class JavaFlumeEventCount { 11 private JavaFlumeEventCount() { 12 } 13 14 public static void main(String[] args) { 15 16 String host = args[0]; 17 int port = Integer.parseInt(args[1]); 18 19 Duration batchInterval = new Duration(Integer.parseInt(args[2])); 20 SparkConf sparkConf = new SparkConf().setAppName("JavaFlumeEventCount"); 21 JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, batchInterval); 22 JavaReceiverInputDStream<SparkFlumeEvent> flumeStream = FlumeUtils.createStream(ssc, host, port); 23 24 flumeStream.count(); 25 26 flumeStream.count().map(new Function<Long, String>() { 27 @Override 28 public String call(Long in) { 29 return "Received " + in + " flume events."; 30 } 31 }).print(); 32 33 ssc.start(); 34 ssc.awaitTermination(); 35 } 36 }
这个和官方的区别是删除了参数个数检查和增加了自定义时间间隔(分割流),也就是第三个参数。这个类并没有做太多处理,入门为主。
三、打包这个类到ifeng_spark.jar,连同spark-streaming-flume_2.10-1.0.1.jar一起上传到spark集群中的节点上。
四、启动flume,这个flume的sink要用avro,指定要发送到的spark集群中的一个节点,我们这里是10.32.21.165:11000。
五、在spark安装根目录下执行如下命令:
./bin/spark-submit --master spark://10.32.21.165:8070 --driver-memory 4G --executor-memory 4G --jars /usr/lib/spark-1.0.0-cdh4/lib/spark-streaming-flume_2.10-1.0.1.jar,/usr/lib/flume-ng-1.4-cdh4.6.0/lib/flume-ng-sdk-1.4.0-cdh6.0.jar /usr/lib/spark-1.0.0-cdh4/ifeng_spark.jar --class com.spark_streaming.JavaFlumeEventCount 10.32.21.165 11000 2000
这个命令中的参数解释请参考下面参考3中的解释,也可以自己增加一些参数,需要注意的是配置内存,自己根据需要自行增加内存(driver、executor)防止OOM。另外jars可以同时加载多个jar包,逗号分隔。记得指定类后需要指定3个参数。
如果没有指定Flume的sdk包,会爆如下错误:
java.lang.NoClassDefFoundError: Lorg/apache/flume/source/avro/AvroFlumeEvent;没有找到类。这个类在flume的sdk包内,在jars参数中指定jar包位置就可以。
还有就是要将自己定义的业务类的jar单独列出,不要放在jars参数指定,否则也会有错误抛出。
运行后可以看到大量的输出信息,然后可以看到有数据的RDD会统计出这个RDD有多少行,截图如下,最后的部分就是这2秒(上面命令最后的参数设定的)统计结果:
至此,flume-ng与spark的对接成功,这只是一个入门实验。可根据需要灵活编写相关的业务类来实现实时处理Flume传输的数据。
spark streaming和一些数据传输工具对接可以达到实时处理的目的。
参考:
1、https://spark.apache.org/docs/0.9.0/streaming-programming-guide.html
2、http://www.cnblogs.com/cenyuhai/p/3577204.html
3、http://blog.csdn.net/book_mmicky/article/details/25714545 , 重要的参数解释
4、http://blog.csdn.net/lskyne/article/details/37561235 , 这是一个例子
5、http://search.maven.org/#search%7Cga%7C1%7Cg%3A%22org.apache.spark%22%20 , spark-flume插件下载
6、http://outofmemory.cn/spark/configuration , spark一些可配置参数说明
7、https://issues.apache.org/jira/browse/SPARK-1916 ,这是1.0.1之前版本中spark streaming与flume对接的一个bug信息
8、https://github.com/apache/spark/tree/master/examples/src/main/java/org/apache/spark/examples/streaming , 这是java版本的spark streaming的一些例子,里面有flume的一个