本篇文章主要介绍了"第1课:SparkStreaming 三板斧之一:解密SparkStreaming另类实验及SparkStreaming本质解析",主要涉及到方面的内容,对于其他数据库感兴趣的同学可以参考一下:
本期内容:SparkStreaming在线另类实验瞬间理解SparkStreaming的本质 SparkStreaming 是Spark Core上的一个子框...
本期内容:
SparkStreaming在线另类实验
瞬间理解SparkStreaming的本质
SparkStreaming 是Spark Core上的一个子框架,如果我们能够完全精通了一个子框架,我们就能够更好的驾驭Spark。SparkStreaming和Spark SQL是目前最流行的框架,从研究角度而言,Spark SQL有太多涉及到SQL优化的问题,不太适应用来深入研究。而SparkStreaming和其他的框架不同,它更像是SparkCore的一个应用程序。如果我们能深入的了解SparkStreaming,那我们就可以写出非常复杂的应用程序。
SparkStreaming的优势是可以结合SparkSQL、图计算、机器学习,功能更加强大。这个时代,单纯的流计算已经无法满足客户的需求啦。在Spark中SparkStreaming也是最容易出现问题的,因为它是不断的运行,内部比较复杂。
本次实验基于如下博客中的程序代码
IMF课程的第94课:SparkStreaming 实现广告计费系统中在线黑名单过滤实战
http://lqding.blog.51cto.com/9123978/1769290
为了更好的查看Job的运行情况,我们启动history-server
root@spark-master:/usr/local/spark/spark-1.6.0-bin-hadoop2.6/sbin# ./start-history-server.sh
history-server启动失败,查看日志报如下信息:
Exception in thread "main" java.lang.reflect.InvocationTargetException
at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
at org.apache.spark.deploy.history.HistoryServer$.main(HistoryServer.scala:235)
at org.apache.spark.deploy.history.HistoryServer.main(HistoryServer.scala)
Caused by: java.lang.IllegalArgumentException: Log directory specified does not exist: file:/tmp/spark-events. Did you configure the correct one through spark.history.fs.logDirectory?
根据报错信息,大致可以看出是Log目录不存在。创建目录
root@spark-master:/tmp# hdfs dfs -mkdir /historyServerForSpark/
配置spark-env.sh,添加一个环境变量,让history server的logDirectory指向上面建立的目录
export SPARK_HISTORY_OPTS="-Dspark.history.fs.logDirectory=hdfs://spark-master:8020/historyServerForSpark"
配置spark-defaults.conf,添加如下配置项:
#是否记录作业产生的事件或者运行状态(job,stage等使用内存等信息)
spark.eventLog.enabled true
#如果记录作业产生的事件或者运行状态,则将事件写入什么位置
spark.eventLog.dir hdfs://spark-master:8020/historyServerForSpark
#http history的监听端口号,通过http://hadoop.master:18080访问
spark.history.ui.port 18080
#Spark history日志位置
park.history.fs.logDirectory=hdfs://spark-master:8020/historyServerForSpark
再次启动history-server,问题解决。
WEB界面如下:

为了可以更清晰的看清楚Streaming运行的各个环节,我们可以通过将batchInterval的值设置的更大。例如5分钟。
将程序上传至spark集群
运行Spark程序
root@spark-master:~# /usr/local/spark/spark-1.6.0-bin-hadoop2.6/bin/spark-submit --class com.dt.spark.streaming.OnlineBlackListFilter --master spark://spark-master:7077 ./spark.jar
打开netcat,并发送一些数据,内容如下:
root@spark-master:~# nc -lk 9999
134343 Hadoop
343434 spark
3432777 Java
0983743 Hbase
893434 Mathou
程序输入结果为: