第87课：Flume推送数据到SparkStreaming案例实战和内幕源码解密

风起漂泊

　　一、Flume推送数据给SparkStreaming,流程图如下：
　　
　　
如上图所示，Flume sink会根据配置的host与port将事件(以AVRO的方式)发送到spark 程序，由spark程序完成处理。然后需要一个Flume的Agent来将数据接收，暂存及转发。这些数据可以来自于本地文件、HDFS等
二、代码实战（Java版本代码）

　　第一步：配置SparkConf：
　　1，至少2条线程：因为Spark Streaming应用程序在运行的时候，至少有一条线程用于不断的循环接收数据，并且至少有一条线程用于处理接受的数据（否则的话无法有线程用于处理数据，随着时间的推移，内存和磁盘都会不堪重负）；
　　2，对于集群而言，每个Executor一般肯定不止一个Thread，那对于处理Spark Streaming的应用程序而言，每个Executor一般分配多少Core比较合适？根据我们过去的经验，5个左右的Core是最佳的（一个Executor分配为奇数个Core表现最佳，例如3个、5个、7个Core等）；
　　

　　SparkConf conf = new SparkConf().setMaster("local[4]").setAppName("FlumePushDataToSparkStreaming");
　　

　　
第二步：创建SparkStreamingContext：
1，这个是SparkStreaming应用程序所有功能的起始点和程序调度的核心SparkStreamingContext的构建可以基于SparkConf参数，也可基于持久化的SparkStreamingContext的内容
来恢复过来（典型的场景是Driver崩溃后重新启动，由于Spark Streaming具有连续7*24小时不间断运行的特征，所有需要在Driver重新启动后继续上一次的状态，此时的状态恢复需要基于曾经的Checkpoint）；
2，在一个Spark Streaming应用程序中可以创建若干个SparkStreamingContext对象，使用下一个SparkStreamingContext之前需要把前面正在运行的SparkStreamingContext对象关闭掉，
由此，我们获得一个重大的启发SparkStreaming框架也只是Spark Core上的一个应用程序而已，只不过Spark Streaming框架箱运行的话需要Spark工程师写业务逻辑处理代码；
　　

　　JavaStreamingContext jsc = newJavaStreamingContext(conf, Durations.seconds(30));



第三步：创建Spark Streaming输入数据来源input Stream：
1，数据输入来源可以基于File、HDFS、Flume、Kafka、Socket等
2, 在这里我们指定数据来源于网络Socket端口，Spark Streaming连接上该端口并在运行的时候一直监听该端口的数据（当然该端口服务首先必须存在）,并且在后续会根据业务需要不断的有数据产生(当然对于Spark Streaming应用程序的运行而言，有无数据其处理流程都是一样的)；
3,如果经常在每间隔5秒钟没有数据的话不断的启动空的Job其实是会造成调度资源的浪费，因为并没有数据需要发生计算，所以实例的企业级生成环境的代码在具体提交Job前会判断是否有数据，如果没有的话就不再提交Job；
JavaReceiverInputDStreamlines = FlumeUtils.createStream(jsc,"master", 9999);    //master为集群Master节点主机名




第四步：接下来就像对于RDD编程一样基于DStream进行编程，原因是DStream是RDD产生的模板（或者说类），在SparkStreaming具体发生计算前，其实质是把每个Batch的DStream的操作翻译成为对RDD的操作，对初始的DStream进行Transformation级别的处理，例如map、filter等高阶函数等的编程，来进行具体的数据计算
第4.1步：讲每一行的字符串拆分成单个的单词
　　JavaDStream words = lines.flatMap(new FlatMapFunction() {
　　//如果是Scala，由于SAM转换，所以可以写成val words = lines.flatMap { line =>line.split(" ")}
　　    @Override
　　    public Iterable call(SparkFlumeEventevent) throws Exception { Stringline = new String(event.event().getBody().array());
　　            return Arrays.asList(line.split(" "));}
　　                   });
　　

第4.2步：在单词拆分的基础上对每个单词实例计数为1，也就是word => (word, 1)
　　
JavaPairDStream pairs = words.mapToPair(new PairFunction() {
     @Override
     publicTuple2 call(String word) throwsException {
          return newTuple2(word, 1);}
                   });
　　第4.3步：在每个单词实例计数为1基础之上统计每个单词在文件中出现的总次数
　　
JavaPairDStream wordsCount = pairs.reduceByKey(new Function2() {
//对相同的Key，进行Value的累计（包括Local和Reducer级别同时Reduce）
    @Override
    publicInteger call(Integer v1, Integer v2) throws Exception {
           return v1 + v2; }
                   });
　　


此处的print并不会直接触发Job的执行，因为现在的一切都是在Spark Streaming框架的控制之下的，对于Spark Streaming而言具体是否触发真正的Job运行是基于设置的Duration时间间隔的
诸位一定要注意的是Spark Streaming应用程序要想执行具体的Job，对DStream就必须有output Stream操作，outputStream有很多类型的函数触发，类print、saveAsTextFile、saveAsHadoopFiles等，最为重要的一个
方法是foraeachRDD,因为Spark Streaming处理的结果一般都会放在Redis、DB、DashBoard等上面，foreachRDD主要就是用用来完成这些功能的，而且可以随意的自定义具体数据到底放在哪里！！！
wordsCount.print();

　　

Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的，当然其内部有消息循环体，用于接受应用程序本身或者Executor中的消息；
jsc.start();
jsc.awaitTermination();
　　jsc.close();
　　

　　三、配置

　　为了实现上述目的，需要将flume配置目录下的flume-conf.propeties配置文件改成如下方式
　　
　　
如上图所示配置后，flume sink会根据配置的host与port将事件发送到spark然后需要一个Flume的Agent来将数据接收，暂存及转发。
四、源码解析

　　如上图所示，通过传入的hostname和端口，Flume通过事件的方式构建DStream流

　　
　　而FlumeInputDStream继承自ReciverInputDStream，FlumeInputDStream通过getReceiver方法接收数据

　　而FlumeReceiver是通管netty的方式进行通讯的
　　

　　五、启动Flume
　　bin/flume-ng agent -n agent1 --conf/usr/local/flume/apache-flume-1.6.0-bin/conf/ -f/usr/local/flume/apache-flume-1.6.0-bin/conf/exec.conf-Dflume.root.logger=DEBUG,console
　　

　　六、将数据拷贝到监控目录，查看程序运行结果

　　

　　程序运行结果如下：

备注：
资料来源于：DT_大数据梦工厂（IMF传奇行动绝密课程）
更多私密内容，请关注微信公众号：DT_Spark
如果您对大数据Spark感兴趣，可以免费听由王家林老师每天晚上20：00开设的Spark永久免费公开课，地址YY房间号：68917580