?

本文将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中的例子修改而来)，代码如下：

?

 package com.spark_streaming;
 
 import org.apache.spark.SparkConf;
 import org.apache.spark.api.java.function.Function;
 import org.apache.spark.streaming.*;
 import org.apache.spark.streaming.api.java.*;
  import org.apache.spark.streaming.flume.FlumeUtils;
  import org.apache.spark.streaming.flume.SparkFlumeEvent;
  
 public final class JavaFlumeEventCount {
   private JavaFlumeEventCount() {
   }
 
   public static void main(String[] args) {
 
     String host = args[0];
     int port = Integer.parseInt(args[1]);
 
     Duration batchInterval = new Duration(Integer.parseInt(args[2]));
     SparkConf sparkConf = new SparkConf().setAppName("JavaFlumeEventCount");
     JavaStreamingContext ssc = new JavaStreamingContext(sparkConf, batchInterval);
     JavaReceiverInputDStream flumeStream = FlumeUtils.createStream(ssc, host, port);
 
     flumeStream.count();
 
     flumeStream.count().map(new Function() {
       @Override
       public String call(Long in) {
         return "Received " + in + " flume events.";
       }
     }).print();
 
     ssc.start();
     ssc.awaitTermination();
  }
 }

?

?

?

这个和官方的区别是删除了参数个数检查和增加了自定义时间间隔(分割流)，也就是第三个参数。这个类并没有做太多处理，入门为主。

?

三、打包这个类到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和一些数据传输工具对接可以达到实时处理的目的。