快乐大数据第11课SparkStream

第一部分：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 

 将处理过的数据输出到外部系统

 内置输出

• print

• 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(&＃8220;local[2]&＃8221;)                       

val ssc = new StreamingContext(conf,Seconds(5))                          流式上下文

val ds = ssc.socketTextStream(&＃8220;192.168.183.100&＃8221;,8888) 流式数据输入

val rs = ds.flatMap(_.split(&＃8221; &＃8220;)).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 &＃8211;create &＃8211;zookeeper 192.168.183.101:2181 &＃8211;replication-factor 1 &＃8211;partitions 3 &＃8211;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;