第一部分：Stream程序设计原理
#SparkStreaming设计动机
很多重要的应用要处理大量在线流式数据，并返回近实时的结果
• 社交网络趋势跟踪
• 电商网站指标统计
• 广告系统
具备分布式流式处理框架的基本特征
• 良好的扩展性
• 低延迟（秒级别）
#什么是SparkStreaming
将Spark扩展为大规模流处理系统
可以扩展到100节点规模，达到秒级延迟
高效且具有良好的容错性
提供了类似批处理的API，很容易实现复杂算法
易用性好
• 提供很多高级算子，实现复杂运算非常简单
• 流式API和批处理API很类似，学习成本低
平台统一
• 不需要维护两套系统分别用于批处理和流式处理
• 可以自由调用Spark的组件，如SparkSQL、Mllib
生态丰富
• 支持各种数据源和数据格式
• 社区活跃，发展迅猛
#SparkStreaming原理
将流式计算转化为一批很小的、确定的批处理作业（micro-batch）
• 以X秒为单位将数据流切分成离散的作业
• 将每批数据看成RDD，使用RDD操作符处理
• 最终结果以RDD为单位返回（写入HDFS或者其他系统）
#spark各组件之间的数据转化
Data Set transformation
Spark Core RDD RDD -> RDD
Spark SQL DataFrame/DataSet DataFrame/DataSet ->DataFrame/DataSet
SparkStreaming DStream Dstream -> DStream
#核心概念-DStream
将连续的数据进行离散表示
DStream中每一个离散的片段都是一个RDD
#Stream Data Source：内置数据源
socketTextStre
textFileStream
其他
外部数据源
• Kafka
• Flume
• ZeroMQ
• 其他
#Stream Transformation
类RDD转换
• map、flatMap、filter、reduce
• groupBykey、reudceByKey、join
Streaming独有转换
• window
• mapWithState
# Stream Output
将处理过的数据输出到外部系统
内置输出
• saveAsTextFiles
自定义输出
• foreachRDD
#基于窗口的transformation函数 window、countByWindow、reduceByWindow等
window length：窗口长度
sliding interval：滑动窗口时间间隔
示例：pairs.reduceByKeyAndWindow((a:Int,b:Int) => (a + b), Seconds(3),
Seconds(2))
#mapWithState
由Spark Streaming自己维护状态信息，不需要借助外部的存储系统
相对updateStateByKey性能提升10倍左右
相对updateStateByKey维护的key状态多10倍
接收参数为StateSpec对象，返回一个新的DStream
#Stream Checkpoint
可以checkpoint的两种类型数据
• Metadata checkpointing，针对Driver中的元数据设置
检查点，包括配置信息、DStream一系列操作、提
交了job但未完成的batch等
• Data checkpointing，保存stateful带状态操作的数据
Checkpoint局限性
• Application 重新编译后，从checkpoint中恢复会失
败，需要清空checkpoint
第二部分：Spark程序设计
#Spark Streaming程序设计
val conf = new SparkConf().setMaster("local[2]")
val ssc = new StreamingContext(conf,Seconds(5)) 流式上下文
val ds = ssc.socketTextStream("192.168.183.100",8888) 流式数据输入
val rs = ds.flatMap(_.split(" ")).map((_,1))
.reduceByKey(_ + _) 流式转换
rs.print() 流式数据输出
ssc.start()
ssc.awaitTermination() 启动流式处理
#例子
#StreamSockerWordCount
程序见StreamSocketWordCount.scala
启动 在node01中
.sparkstreaming监听端口计算wordcount
安装nc：yum install -y nc
启动一个socket服务端口：nc -lk 8888
在node01的控制台
hello world
hello socket
hello spark
hello name
hello spark
#StreamWindowWordCount
#演示mapwithState
#用户行为数据分析
（一）在kafka上创建主题
在node03上，
bin/kafka-topics.sh --create --zookeeper 192.168.183.101:2181 --replication-factor 1 --partitions 3 --topic DeviceEvents
（二）使用mysql，创建表
在node02上，
mysql -h192.168.183.102 -uhive -phive123
create database streamdb;
use streamdb;
CREATE TABLE `device_click` (
`id` int(11) NOT NULL AUTO_INCREMENT,
`device_id` varchar(512) DEFAULT NULL,
`device_type` varchar(512) DEFAULT NULL,
`time` varchar(512) DEFAULT NULL,
`click_count` int(11) NOT NULL,
PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;
（三）往Kafka里面写数据
（四）写一个Stream程序，来读这个程序StreamDeviceClickCountAnalytics
在node2上启动mysql后
select * from device_click;
select device_type,sum(click_count) from device_click group by device_type;
网友评论