IMF传奇行动第85课:Spark Streaming第四课:基于HDFS的Spark Streaming案例实战和内幕源码解密
本课分三部分讲解:
第一部分准备HDFS和Spark集群环境;
第二部分用Java代码编写基于HDFS的Spark Streaming案例;
第三部分Spark Streaming on HDFS源码解密。
一、准备HDFS和Spark集群环境
1. 启动HDFS,如下图所示:
# start-dfs.sh
2. 启动Spark集群,如下图所示:
# cd $SPARK_HOME/bin
# ./start-all.sh
3. 启动Spark History Server,如下图所示:
# start-history-server.sh
二、案例代码实战
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
此时SparkStreaming会每隔5秒执行一次,不断的扫描监控目录下是否有新的文件。
d. 上传文件到HDFS 中的Data 目录下
# cd $HADOOP_HOME
# hadoop dfs -put ./CHANGES.txt /library/SparkStreaming/Data
e. 输出结果
三、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) }
思考为什么会报下面的错误?
原因:
- Streaming会定期的进行checkpoint
- 重新启动程序的时候,它会进入曾经checkpoint的目录中。如果没有做额外配置的时候,所有的信息都会放在checkpoint的目录中(包括曾经的应用程序信息),因此下次再次启动的时候就会报错,导致无法初始化ShuffledDStream。
课程笔记来源:
DT大数据梦工厂IMF传奇行动课程学员整理。YY直播永久课堂频道68917580每晚8点准时开课。
Life is short, you need spark!
IMF传奇行动第85课:Spark Streaming第四课:基于HDFS的Spark Streaming案例实战和内幕源码解密相关推荐
- 第91课:SparkStreaming基于Kafka Direct案例实战和内幕源码解密 java.lang.ClassNotFoundException 踩坑解决问题详细内幕版本
第91课:SparkStreaming基于Kafka Direct案例实战和内幕源码解密 /* * *王家林老师授课http://weibo.com/ilovepains */ 每天晚上20: ...
- 第93课:SparkStreaming updateStateByKey 基本操作综合案例实战和内幕源码解密
Spark Streaming的DStream为我们提供了一个updateStateByKey方法,它的主要功能是可以随着时间的流逝在Spark Streaming中为每一个key维护一份state状 ...
- 《Spark SQL大数据实例开发》9.2 综合案例实战——电商网站搜索排名统计
<Spark SQL大数据实例开发>9.2 综合案例实战--电商网站搜索排名统计 9.2.1 案例概述 本节演示一个网站搜索综合案例:以京东为例,用户登录京东网站,在搜索栏中输入搜 ...
- 深入理解Spark 2.1 Core (十):Shuffle Map 端的原理与源码分析
在上一篇<深入理解Spark 2.1 Core (九):迭代计算和Shuffle的原理与源码分析>提到经过迭代计算后, SortShuffleWriter.write中: // 根据排序方 ...
- 深入理解Spark 2.1 Core (五):Standalone模式运行的原理与源码分析
概述 前几篇博文都在介绍Spark的调度,这篇博文我们从更加宏观的调度看Spark,讲讲Spark的部署模式.Spark部署模式分以下几种: local 模式 local-cluster 模式 Sta ...
- 深入理解Spark 2.1 Core (十一):Shuffle Reduce 端的原理与源码分析
我们曾经在<深入理解Spark 2.1 Core (一):RDD的原理与源码分析 >讲解过: 为了有效地实现容错,RDD提供了一种高度受限的共享内存,即RDD是只读的,并且只能通过其他RD ...
- 深入理解Spark 2.1 Core (九):迭代计算和Shuffle的原理与源码分析
Task有两个子类,一个是非最后的Stage的Task,ShuffleMapTask:一个是最后的Stage的Task,ResultTask.它们都覆盖了Task的runTask方法. 我们来看一下S ...
- 深入理解Spark 2.1 Core (六):Standalone模式运行的原理与源码分析
我们讲到了如何启动Master和Worker,还讲到了如何回收资源.但是,我们没有将AppClient是如何启动的,其实它们的启动也涉及到了资源是如何调度的.这篇博文,我们就来讲一下AppClient ...
- 深入理解Spark 2.1 Core (二):DAG调度器的原理与源码分析
概述 上一篇<深入理解Spark(一):RDD实现及源码分析 >提到: 定义RDD之后,程序员就可以在动作(注:即action操作)中使用RDD了.动作是向应用程序返回值,或向存储系统导出 ...
- 深入理解Spark 2.1 Core (四):运算结果处理和容错的原理与源码分析
这篇博文,我们来看看当executor计算完任务后, Spark是如何处理获取的计算结果与容错的. 调用栈如下: TaskSchedulerImpl.statusUpdate TaskResultGe ...
最新文章
- mysql2008jar包下载_求java jdbc 连sql server2008的jar包
- 支付宝分库分表中间件--zdal简介
- 【LeetCode】732. 我的日程安排表 III
- hihocoder234周 计算不包含黑点的矩形个数
- c++计算字符串中浮点数
- IBM Rhapsody 报错找不到 VC11 目录的解决方法
- ssm框架整合以及登录案例
- c语言定时器1khz占空比,555定时器产生方波原理(四款555定时器产生方波的电路详解)...
- Comic Sc​​roller - 将漫画网站中一话整理一整页的插件
- GNN教程:采样模型GraphSage理解(工业上图模型的雏形)
- linux把大文件分成小文件,linux文件分割(将大的日志文件分割成小的)
- python爬取公众号阅读量_公众号提升阅读量!免费推荐几个互阅推广平台.
- R语言ggplot2可视化在轴标签中添加上标(Superscript)和下标(subscript)实战
- 梆梆加固的病毒分析-破解篇
- 多目标跟踪算法中之图匹配——匈牙利算法和KM算法详解
- 代理记账公司是如何进行缴税的?
- 前端技能树,面试复习第 1 天—— 建立前端知识框架 | HTML 知识考察点
- 我只认两种产业互联网形态
- c语言二进制数以字符形式输出,如何用C语言输出二进制数据
- Nsight System的安装和使用
热门文章
- Linux下编译程序/usr/bin/ld: cannot find -l*错误的解决方法
- 微信公众号调取相册和摄像头功能,实现图片上传
- TestCenter测试管理工具功能详解十二(Q)
- 制作mac系统引导盘
- python采用强制缩进的方式使得代码具有极佳的可读性_计算机专业高级语言程序设计课程改革探索...
- 键盘按键开关种类简介
- lamp技术研发人员的必备
- lua菜鸟教程_lua经典学习教程.pdf
- 1076: 三位数求解 C语言
- Oracle Becomes 2nd Largest Software Company