通过案例对spark streaming透彻理解三板斧之一
2016-05-02 23:15
447 查看
简介: 王家林:DT大数据梦工厂创始人和首席专家.
联系邮箱18610086859@126.com
电话:18610086859 QQ:1740415547 微信号:18610086859
spark的核心是spark core,spark streaming,spark graph,spark mlib 其实是构建在spark core之上的一个应用程序,如果要构建一个强大的spark应用程序 ,spark streaming 是一个值得借鉴的参考,spark streaming涉及多个job交叉配合,里面涉及到了spark的所有的核心组件,如果对spark streaming 精通了的话,可以说就精通了整个spark,所以精通掌握spark streaming是至关重要的。
spark另类实现。在试验中。通过调节放大Batch Interval的方式300s,观察数据流入流出。
object OnlineBlackListFilter {
def main(args: Array[String]){
/**
* 第1步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,
* 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
* 为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如
* 只有1G的内存)的初学者 *
*/
val conf = new SparkConf() //创建SparkConf对象
conf.setAppName("OnlineBlackListFilter") //设置应用程序的名称,在程序运行的监控界面可以看到名称
conf.setMaster("spark://Master:7077") //此时,程序在Spark集群
val ssc = new StreamingContext(conf, Seconds(300))
/**
* 黑名单数据准备,实际上黑名单一般都是动态的,例如在Redis或者数据库中,黑名单的生成往往有复杂的业务
* 逻辑,具体情况算法不同,但是在Spark Streaming进行处理的时候每次都能工访问完整的信息
*/
val blackList = Array(("hadoop", true),("mahout", true))
val blackListRDD = ssc.sparkContext.parallelize(blackList, 8)
val adsClickStream = ssc.socketTextStream("Master", 9999)
/**
* 此处模拟的广告点击的每条数据的格式为:time、name
* 此处map操作的结果是name、(time,name)的格式
*/
val adsClickStreamFormatted = adsClickStream.map { ads => (ads.split(" ")(1), ads) }
adsClickStreamFormatted.transform(userClickRDD => {
//通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容,又获得了相应点击内容是否在黑名单中
val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)
/**
* 进行filter过滤的时候,其输入元素是一个Tuple:(name,((time,name), boolean))
* 其中第一个元素是黑名单的名称,第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值
* 如果存在的话,表面当前广告点击是黑名单,需要过滤掉,否则的话则是有效点击内容;
*/
val validClicked = joinedBlackListRDD.filter(joinedItem => {
if(joinedItem._2._2.getOrElse(false))
{
false
} else {
true
}
})
validClicked.map(validClick => {validClick._2._1})
}).print
/**
* 计算后的有效数据一般都会写入Kafka中,下游的计费系统会从kafka中pull到有效数据进行计费
*/
ssc.start()
ssc.awaitTermination()
}
}
打开9999端口 nc -lk 9999
输入测试数据
2255554 Spark
455554444 Hadoop
55555 Flink
66666 Kafka
6666855 RockySpark
666638 Scala
66666 DT_Spark
在浏览器history_server中查看
点击最上面的任务后
总共竟然有5个Job。
job0主要用于负载均衡
job1的时间是1.5min, 从各个job所花的时间来看,整个应用运行的时间大部分时间是花在了receiver上
job2的时间
Spark Streaming的一个应用中,运行了这么多Job
联系邮箱18610086859@126.com
电话:18610086859 QQ:1740415547 微信号:18610086859
spark的核心是spark core,spark streaming,spark graph,spark mlib 其实是构建在spark core之上的一个应用程序,如果要构建一个强大的spark应用程序 ,spark streaming 是一个值得借鉴的参考,spark streaming涉及多个job交叉配合,里面涉及到了spark的所有的核心组件,如果对spark streaming 精通了的话,可以说就精通了整个spark,所以精通掌握spark streaming是至关重要的。
spark另类实现。在试验中。通过调节放大Batch Interval的方式300s,观察数据流入流出。
object OnlineBlackListFilter {
def main(args: Array[String]){
/**
* 第1步:创建Spark的配置对象SparkConf,设置Spark程序的运行时的配置信息,
* 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
* 为local,则代表Spark程序在本地运行,特别适合于机器配置条件非常差(例如
* 只有1G的内存)的初学者 *
*/
val conf = new SparkConf() //创建SparkConf对象
conf.setAppName("OnlineBlackListFilter") //设置应用程序的名称,在程序运行的监控界面可以看到名称
conf.setMaster("spark://Master:7077") //此时,程序在Spark集群
val ssc = new StreamingContext(conf, Seconds(300))
/**
* 黑名单数据准备,实际上黑名单一般都是动态的,例如在Redis或者数据库中,黑名单的生成往往有复杂的业务
* 逻辑,具体情况算法不同,但是在Spark Streaming进行处理的时候每次都能工访问完整的信息
*/
val blackList = Array(("hadoop", true),("mahout", true))
val blackListRDD = ssc.sparkContext.parallelize(blackList, 8)
val adsClickStream = ssc.socketTextStream("Master", 9999)
/**
* 此处模拟的广告点击的每条数据的格式为:time、name
* 此处map操作的结果是name、(time,name)的格式
*/
val adsClickStreamFormatted = adsClickStream.map { ads => (ads.split(" ")(1), ads) }
adsClickStreamFormatted.transform(userClickRDD => {
//通过leftOuterJoin操作既保留了左侧用户广告点击内容的RDD的所有内容,又获得了相应点击内容是否在黑名单中
val joinedBlackListRDD = userClickRDD.leftOuterJoin(blackListRDD)
/**
* 进行filter过滤的时候,其输入元素是一个Tuple:(name,((time,name), boolean))
* 其中第一个元素是黑名单的名称,第二元素的第二个元素是进行leftOuterJoin的时候是否存在在值
* 如果存在的话,表面当前广告点击是黑名单,需要过滤掉,否则的话则是有效点击内容;
*/
val validClicked = joinedBlackListRDD.filter(joinedItem => {
if(joinedItem._2._2.getOrElse(false))
{
false
} else {
true
}
})
validClicked.map(validClick => {validClick._2._1})
/**
* 计算后的有效数据一般都会写入Kafka中,下游的计费系统会从kafka中pull到有效数据进行计费
*/
ssc.start()
ssc.awaitTermination()
}
}
打开9999端口 nc -lk 9999
输入测试数据
2255554 Spark
455554444 Hadoop
55555 Flink
66666 Kafka
6666855 RockySpark
666638 Scala
66666 DT_Spark
在浏览器history_server中查看
点击最上面的任务后
总共竟然有5个Job。
job0主要用于负载均衡
job1的时间是1.5min, 从各个job所花的时间来看,整个应用运行的时间大部分时间是花在了receiver上
job2的时间
Spark Streaming的一个应用中,运行了这么多Job
相关文章推荐
- 新书到手 TRANSACTION PROCESSING:CONCEPTS AND TECHNIQUES
- 关于启明星系统(OA系统,预定系统,请假系统等)安全性的说明
- Java并发实现一(并发的实现之Thread和Runnable的区别)
- 高吞吐量的分布式发布订阅消息系统Kafka-- 管理工具 Kafka Manager
- 图片嵌入隐藏-大容量的信息隐藏算法
- A Tour of Golang (二)
- bzoj 1101: [POI2007]Zap
- 【裸奔吧linux】netstat: 显示网络连接,路由表,接口统计信息,多播成员
- Precision,Recall和F1
- java对redis的操作
- SDUT 2168 Mathmen(贪心)
- linux 中 alias 命令用法
- Codeforces Round #348 (VK Cup 2016 Round 2, Div. 2 Edition)-B. Little Artem and Grasshopper(模拟)
- 完全卸载mysql数据库
- leetcode_096 Unique Binary Search Trees
- 宽字符和字符
- SQL语句命令
- OC里面的类类型
- 10周机电大作业设计
- 【翻译】使用nginx作为反向代理服务器,uWSGI作为应用服务器来部署flask应用