IMF传奇行动第85课：Spark Streaming第四课：基于HDFS的Spark Streaming案例实战和内幕源码解密

超、凢脫俗 2022-08-21 12:42 98阅读 0赞

本课分三部分讲解：

第一部分准备HDFS和Spark集群环境；

第二部分用Java代码编写基于HDFS的Spark Streaming案例；

第三部分Spark Streaming on HDFS源码解密。

** **

# 一、准备HDFS和Spark集群环境 #

### 1. 启动HDFS，如下图所示： ###

\# start-dfs.sh

![SouthEast][]

### 2. 启动Spark集群，如下图所示： ###

\# cd $SPARK\_HOME/bin

\# ./start-all.sh

![SouthEast 1][]

### 3. 启动Spark History Server，如下图所示： ###

\# start-history-server.sh

# ![SouthEast 2][]    #

#     #

# 二、案例代码实战 #

### 1. 用Java代码编写基于HDFS的SparkStreaming案例： ###

package  com.dt.spark.SparkApps.sparkstreaming;  
import org.apache.spark.SparkConf;  
import org.apache.spark.SparkContext;  
import org.apache.spark.api.java.function.FlatMapFunction;  
import org.apache.spark.api.java.function.Function2;  
import org.apache.spark.api.java.function.PairFunction;  
import org.apache.spark.streaming.Durations;  
import org.apache.spark.streaming.api.java.\*;  
import scala.Tuple2;  
import java.util.Arrays;  
*/\*\*  
 \* @author DT大数据梦工厂*

* \*新浪微博：http://weibo.com/ilovepains/  
 \*/*  
public class SparkStreamingOnHDFS \{  
    publicstatic void main(String\[\]args)\{  
        */\*\*  
         \* 第一步：配置SparkConf  
         \* 1. 至少两条线程：  
         \* 因为SparkStreaming应用程序在运行的时候，至少有一条线程用于不断的循环接收数据，  
         \* 并且至少有一条线程用于处理接收的数据（否则的话无法有线程用于处理数据，随着时间的推移，内存和磁盘都不堪重负）  
         \* 2. 对于集群而言，每个Executor一般而言肯定不止一个线程，对于处理Spark Streaming的应用程序而言，每个Executor一般分配多少个Core合适呢？根据我们过去的经验，5个左右的core是最佳的（分配为奇数个Core为最佳）。  
         \*/  
       * final SparkConf conf =new SparkConf().setMaster("spark://Master:7077").setAppName("SparkOnStreamingOnHDFS");  
        */\*\*  
         \* 第二步：创建SparkStreamingContext，这个是Spark Streaming应用程序所有功能的起始点和程序调度的核心  
         \* 1，SparkStreamingContext的构建可以基于SparkConf参数，也可以基于持久化SparkStreamingContext的内容来恢复过来(典型的场景是Driver崩溃后重新启动，由于Spark Streaming具有连续7\*24小时不间断运行的特征，所有需要在Driver重新启动后继续上一次的状态，此时状态的恢复需要基于曾经的checkpoint)  
         \* 2，在一个SparkStreaming应用程序中可以创建若干个SparkStreamingContext对象，使用下一个SparkStreamingContext之前需要把前面正在运行的SparkStreamingContext对象关闭掉，由此，我们获得一个重大启发：SparkStreamingContext是Sparkcore上的一个应用程序而已，只不过Spark Streaming框架箱运行的话需要Spark工程师写业务逻辑  
         \*/*  
//       JavaStreamingContext jsc = new JavaStreamingContext(conf,Durations.seconds(5));//Durations.seconds(5)设置每隔5秒  
        final String checkpointDirectory = "hdfs://Master:9000/library/SparkStreaming/Checkpoint\_Data";  
        JavaStreamingContextFactory factory = new  JavaStreamingContextFactory() \{  
            @Override  
            public JavaStreamingContextcreate() \{  
                return *createContext*(checkpointDirectory,conf);  
            \}  
        \};  
        */\*\*  
         \* 可以从失败中恢复Driver，不过还需要制定Driver这个进程运行在Cluster，并且提交应用程序的时候  
         \* 指定\--supervise;  
         \*/  
       * JavaStreamingContext jsc = JavaStreamingContext.*getOrCreate*(checkpointDirectory, factory);  
        */\*\*  
         \* 现在是监控一个文件系统的目录  
         \* 此处没有Receiver，Spark Streaming应用程序只是按照时间间隔监控目录下每个Batch新增的内容(把新增的)作为RDD的数据来源生成原始的RDD  
         \*/  
       * //指定从HDFS中监控的目录  
        JavaDStream lines = jsc.textFileStream("hdfs://Master:9000/library/SparkStreaming/Data");  
        */\*\*  
         \* 第四步：接下来就像对于RDD编程一样基于DStreaming进行编程！！！  
         \* 原因是：  
         \*  DStreaming是RDD产生的模板（或者说类）。  
         \*  在Spark Streaming具体发生计算前其实质是把每个batch的DStream的操作翻译成对RDD的操作！！  
         \*  对初始的DStream进行Transformation级别的处理，例如Map,filter等高阶函数的编程，来进行具体的数据计算。  
         \*  第4.1步：将每一行的字符串拆分成单个单词  
         \*/  
       * JavaDStream<String> words = lines.flatMap(new FlatMapFunction<String,String>() \{  
            public Iterable<String>call(Stringline) throws  Exception \{  
                return Arrays.*asList*(line.split(" "));  
            \}  
        \});  
        */\*\*  
         \* 第4.2步：对初始的JavaRDD进行Transformation级别的处理，例如map，filter等高阶函数等的编程，来进行具体的数据计算  
         \* 在4.1的基础上，在单词拆分的基础上对每个单词实例计数为1，也就是word =>(word,1)  
         \*/  
       * JavaPairDStream<String,Integer> pairs  = words.mapToPair(new PairFunction<String, String, Integer>() \{  
            public Tuple2<String, Integer> call(String word) throws  Exception \{  
                return new Tuple2<String,Integer>(word,1);  
            \}  
        \});  
        */\*\*  
         \* 第4.3步：在每个单词实例计数的基础上统计每个单词在文件中出现的总次数  
         \*/  
       * JavaPairDStream<String,Integer> wordscount =pairs.reduceByKey(new Function2<Integer, Integer, Integer>() \{  
            public Integer call(Integer v1, Integer v2) throws  Exception \{  
                return v1 + v2;  
            \}  
        \});  
        */\*\*  
         \* 此处的print并不会直接触发Job的执行，因为现在的一切都是在SparkStreaming框架控制下的，对于Spark而言具体是否  
         \* 触发真正的Job运行是基于设置的Duration时间间隔的  
         \* 一定要注意的是：SparkStreaming应用程序要想执行具体的Job，对DStream就必须有output Stream操作，  
         \* output Stream有很多类型的函数触发，例如：print，saveAsTextFile,saveAsHadoopFiles等，其实最为重要的一个方法是  
         \* foraeachRDD，因为SparkStreaming处理的结果一般都会放在Redis，DB，DashBoard等上面，foreachRDD主要就是用来完成这些  
         \* 功能的，而且可以随意的自定义具体数据到底存放在哪里！！！  
         \*/  
       * wordscount.print();  
        */\*\*  
         \* Spark Streaming执行引擎也就是Driver开始运行，Driver启动的时候是位于一条新的线程中的。  
         \* 当然其内部有消息循环体用于接收应用程序本身或者Executor的消息；  
         \*/  
       * jsc.start();  
        jsc.awaitTermination();  
        jsc.close();  
    \}  
    */\*\*  
     \* 工厂化模式构建JavaStreamingContext  
     \*/  
   * private static JavaStreamingContextcreateContext(String checkpointDirectory,SparkConf conf)\{  
        System.*out*.println("Creating new context");  
        JavaStreamingContext ssc = new  JavaStreamingContext(conf,Durations.*seconds*(5));  
        ssc.checkpoint(checkpointDirectory);  
        return ssc;  
    \}  
\}

### 2. 在之前准备的Spark集群中运行上面的代码测试效果： ###

a. 创建必要的目录

\# hadoop dfs -mkdir /library/SparkStreaming

\# hadoop dfs -mkdir /library/SparkStreaming/CheckPoint\_Data

\# hadoop dfs -mkdir /library/SparkStreaming/Data

b. 新建一个脚本SparkAppsStreaming.sh ，内容如下

/usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit--class cn.spark.study.SparkStreamigOnHDFS --master spark://Master:7077 /root/Documents/SparkApps/spark-study-java-0.0.1-SNAPSHOT-jar-with-dependencies.jar

c. 运行脚本

\# ./SparkAppsStreaming.sh

![SouthEast 3][]

此时SparkStreaming会每隔5秒执行一次，不断的扫描监控目录下是否有新的文件。

![SouthEast 4][]

d. 上传文件到HDFS 中的Data 目录下

\# cd $HADOOP\_HOME

\# hadoop dfs -put ./CHANGES.txt /library/SparkStreaming/Data

![SouthEast 5][]

e. 输出结果

**![SouthEast 6][]**

** **

# 三、SparkStreaming on HDFS源码解密 #

### 1. JavaStreamingContextFactory的create方法可以创建JavaStreamingContext。 ###

### 2. 而我们在具体实现的时候覆写了该方法，内部就是调用createContext方法来具体实现。上述实战案例中我们实现了createContext方法。 ###

/**  * Factory interface for creating a new JavaStreamingContext  */ trait JavaStreamingContextFactory {   def create(): JavaStreamingContext }

### 3. checkpoint: ###

*  一方面：保持容错
 *  另一方面：保持状态

在开始和结束的时候，每个batch都会进行checkpoint。

/**  * Sets the context to periodically checkpoint the DStream operations for master  * fault-tolerance. The graph will be checkpointed every batch interval.  * @param directory HDFS-compatible directory where the checkpoint data will be reliably stored  */ def checkpoint(directory: String) {   ssc.checkpoint(directory) }

### 4. remember: ###

流式处理中过一段时间数据就会被清理掉，但是可以通过remember可以延长数据在程序中的生命周期，另外延长RDD更长的时间。

**应用场景：**假设数据流进来，进行ML或者GraphX的时候有时需要很长时间，但是bacth定时定条件的清除RDD，所以就可以通过remember使得数据可以延长更长时间。

/**  * Sets each DStreams in this context to remember RDDs it generated in the last given duration.  * DStreams remember RDDs only for a limited duration of duration and releases them for garbage  * collection. This method allows the developer to specify how long to remember the RDDs (  * if the developer wishes to query old data outside the DStream computation).  * @param duration Minimum duration that each DStream should remember its RDDs  */ def remember(duration: Duration) {   ssc.remember(duration) }

### 5. 在JavaStreamingContext中，getOrCreate方法源码如下： ###

如果设置了checkpoint，重启程序的时候，getOrCreate()会重新从checkpoint目录中初始化出StreamingContext。

/**  * Either recreate a StreamingContext from checkpoint data or create a new StreamingContext.  * If checkpoint data exists in the provided `checkpointPath`, then StreamingContext will be  * recreated from the checkpoint data. If the data does not exist, then the provided factory  * will be used to create a JavaStreamingContext.  *  * @param checkpointPath Checkpoint directory used in an earlier JavaStreamingContext program  * @param factory        JavaStreamingContextFactory object to create a new JavaStreamingContext  * @deprecated As of 1.4.0, replaced by `getOrCreate` without JavaStreamingContextFactor.  */ @deprecated("use getOrCreate without JavaStreamingContextFactor", "1.4.0") def getOrCreate(     checkpointPath: String,     factory: JavaStreamingContextFactory   ): JavaStreamingContext = {   val ssc = StreamingContext.getOrCreate(checkpointPath, () => {     factory.create.ssc   })   new JavaStreamingContext(ssc) }

思考为什么会报下面的错误？

![SouthEast 7][]

原因：

*  Streaming会定期的进行checkpoint
 *  重新启动程序的时候，它会进入曾经checkpoint的目录中。如果没有做额外配置的时候，所有的信息都会放在checkpoint的目录中（包括曾经的应用程序信息），因此下次再次启动的时候就会报错，导致无法初始化ShuffledDStream。

课程笔记来源:

DT大数据梦工厂IMF传奇行动课程学员整理。YY直播永久课堂频道68917580每晚8点准时开课。

**Life is short, you need spark!**

![SouthEast 8][]

[SouthEast]: /images/20220724/0dfa5541de28492781945d7198a0c501.png
[SouthEast 1]: /images/20220724/e9ceb01e1b5e4759829e95b465f072ce.png
[SouthEast 2]: /images/20220724/99868896060b49c4971dbec474e8e861.png
[SouthEast 3]: /images/20220724/356d63ad93db401fa6fad4c9989d134a.png
[SouthEast 4]: /images/20220724/e3b40f93abd54a8989b737b846d95524.png
[SouthEast 5]: /images/20220724/ed33416da8c941e9ac1e658f58e96cc7.png
[SouthEast 6]: /images/20220724/3ff75406b23740e798ea74cf3d62e6a0.png
[SouthEast 7]: /images/20220724/961c16cc1e034ebb8ed9af1ad81dfc98.png
[SouthEast 8]: /images/20220724/56977e39e5d44d3995658ee221740d43.png