尚硅谷大数据技术-教程-学习路线-笔记汇总表【课程资料下载】
视频地址：尚硅谷大数据Spark教程从入门到精通_哔哩哔哩_bilibili

尚硅谷大数据技术Spark教程-笔记01【SparkCore（概述、快速上手、运行环境、运行架构）】
尚硅谷大数据技术Spark教程-笔记02【SparkCore（核心编程，RDD-核心属性-执行原理-基础编程-并行度与分区-转换算子）】
尚硅谷大数据技术Spark教程-笔记03【SparkCore（核心编程，RDD-转换算子-案例实操）】
尚硅谷大数据技术Spark教程-笔记04【SparkCore（核心编程，RDD-行动算子-序列化-依赖关系-持久化-分区器-文件读取与保存）】
尚硅谷大数据技术Spark教程-笔记05【SparkCore（核心编程，累加器、广播变量）】
尚硅谷大数据技术Spark教程-笔记06【SparkCore（案例实操，电商网站）】
尚硅谷大数据技术Spark教程-笔记07【Spark内核&源码（环境准备、通信环境、应用程序执行、shuffle、内存管理）】
尚硅谷大数据技术Spark教程-笔记08【SparkSQL（介绍、特点、数据模型、核心编程、案例实操、总结）】
尚硅谷大数据技术Spark教程-笔记09【SparkStreaming（概念、入门、DStream入门、案例实操、总结）】

03_尚硅谷大数据技术之SparkStreaming.pdf

P185【185.尚硅谷_SparkStreaming - 概念 - 介绍】09:25

第1章 SparkStreaming概述

P186【186.尚硅谷_SparkStreaming - 概念 - 原理 & 特点】10:24

第2章 Dstream入门

P187【187.尚硅谷_SparkStreaming - 入门 - WordCount - 实现】14:40

P188【188.尚硅谷_SparkStreaming - 入门 - WordCount - 解析】03:11

第3章 DStream创建

P189【189.尚硅谷_SparkStreaming - DStream创建 - Queue】02:39

P190【190.尚硅谷_SparkStreaming - DStream创建 - 自定义数据采集器】07:36

P191【191.尚硅谷_SparkStreaming - DStream创建 - Socket数据采集器源码解读】03:26

P192【192.尚硅谷_SparkStreaming - DStream创建 - Kafka数据源】10:51

第4章 DStream转换

P193【193.尚硅谷_SparkStreaming - DStream转换 - 状态操作】16:09

P194【194.尚硅谷_SparkStreaming - DStream转换 - 无状态操作 - transform】09:06

P195【195.尚硅谷_SparkStreaming - DStream转换 - 无状态操作 - join】03:59

P196【196.尚硅谷_SparkStreaming - DStream转换 - 有状态操作 - window】12:17

P197【197.尚硅谷_SparkStreaming - DStream转换 - 有状态操作 - window - 补充】08:39

第5章 DStream输出

P198【198.尚硅谷_SparkStreaming - DStream输出】04:43

第6章优雅关闭

P199【199.尚硅谷_SparkStreaming - 优雅地关闭】15:45

P200【200.尚硅谷_SparkStreaming - 优雅地关闭 - 恢复数据】03:30

第7章 SparkStreaming案例实操

P201【201.尚硅谷_SparkStreaming - 案例实操 - 环境和数据准备】16:43

P202【202.尚硅谷_SparkStreaming - 案例实操 - 需求一 - 分析】10:20

P203【203.尚硅谷_SparkStreaming - 案例实操 - 需求一 - 功能实现 - 黑名单判断】19:28

P204【204.尚硅谷_SparkStreaming - 案例实操 - 需求一 - 功能实现 - 统计数据更新】16:26

P205【205.尚硅谷_SparkStreaming - 案例实操 - 需求一 - 功能实现 - 测试 & 简化 & 优化】19:30

P206【206.尚硅谷_SparkStreaming - 案例实操 - 需求二 - 功能实现】09:26

P207【207.尚硅谷_SparkStreaming - 案例实操 - 需求二 - 乱码问题】06:11

P208【208.尚硅谷_SparkStreaming - 案例实操 - 需求三 - 介绍 & 功能实现】15:51

P209【209.尚硅谷_SparkStreaming - 案例实操 - 需求三 - 效果演示】09:54

P210【210.尚硅谷_SparkStreaming - 总结 - 课件梳理】08:12

03_尚硅谷大数据技术之SparkStreaming.pdf

P185【185.尚硅谷_SparkStreaming - 概念 - 介绍】09:25

//数据处理的方式角度
流式（streaming）
数据处理批量（batch）数据处理//数据处理延迟的长短
实时数据处理：毫秒级别
离线数据处理：小时or天 级别Sparkstreaming：准实时（秒，分钟），微批次（时间）的数据处理框架。

第1章 SparkStreaming概述

P186【186.尚硅谷_SparkStreaming - 概念 - 原理 & 特点】10:24

第1章 SparkStreaming概述

1.1 Spark Streaming 是什么

Spark Streaming 用于流式数据的处理。Spark Streaming 支持的数据输入源很多，例如：Kafka、 Flume、Twitter、ZeroMQ 和简单的 TCP 套接字等等。数据输入后可以用 Spark 的高度抽象原语，如：map、reduce、join、window 等进行运算，而结果也能保存在很多地方，如 HDFS，数据库等。

第2章 Dstream入门

P187【187.尚硅谷_SparkStreaming - 入门 - WordCount - 实现】14:40

第 2 章 Dstream 入门

2.1 WordCount 案例实操

package com.atguigu.bigdata.spark.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming01_WordCount {def main(args: Array[String]): Unit = {// TODO 创建环境对象// StreamingContext创建时，需要传递两个参数// 第一个参数表示环境配置val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")// 第二个参数表示批量处理的周期（采集周期）val ssc = new StreamingContext(sparkConf, Seconds(3))// TODO 逻辑处理// 获取端口数据val lines: ReceiverInputDStream[String] = ssc.socketTextStream("localhost", 9999)val words = lines.flatMap(_.split(" "))val wordToOne = words.map((_, 1))val wordToCount: DStream[(String, Int)] = wordToOne.reduceByKey(_ + _)wordToCount.print()// TODO 关闭环境// 由于SparkStreaming采集器是长期执行的任务，所以不能直接关闭。// 如果main方法执行完毕，应用程序也会自动结束，所以不能让main执行完毕。//ssc.stop()// 1. 启动采集器ssc.start()// 2. 等待采集器的关闭ssc.awaitTermination()}
}

P188【188.尚硅谷_SparkStreaming - 入门 - WordCount - 解析】03:11

2.2 WordCount解析

Discretized Stream 是 Spark Streaming 的基础抽象，代表持续性的数据流和经过各种 Spark 原语操作后的结果数据流。在内部实现上，DStream 是一系列连续的 RDD 来表示。每个 RDD 含有一段时间间隔内的数据。

第3章 DStream创建

P189【189.尚硅谷_SparkStreaming - DStream创建 - Queue】02:39

第 3 章 DStream 创建

3.1 RDD 队列

3.1.1 用法及说明

测试过程中，可以通过使用 ssc.queueStream(queueOfRDDs)来创建 DStream，每一个推送到这个队列中的 RDD，都会作为一个 DStream 处理。

3.1.2 案例实操

➢ 需求：循环创建几个 RDD，将 RDD 放入队列。通过 SparkStream 创建 Dstream，计算 WordCount。

package com.atguigu.bigdata.spark.streamingimport org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}import scala.collection.mutableobject SparkStreaming02_Queue {def main(args: Array[String]): Unit = {// TODO 创建环境对象// StreamingContext创建时，需要传递两个参数// 第一个参数表示环境配置val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")// 第二个参数表示批量处理的周期（采集周期）val ssc = new StreamingContext(sparkConf, Seconds(3))val rddQueue = new mutable.Queue[RDD[Int]]()val inputStream = ssc.queueStream(rddQueue, oneAtATime = false)val mappedStream = inputStream.map((_, 1))val reducedStream = mappedStream.reduceByKey(_ + _)reducedStream.print()ssc.start()for (i <- 1 to 5) {rddQueue += ssc.sparkContext.makeRDD(1 to 300, 10)Thread.sleep(2000)}ssc.awaitTermination()}
}

P190【190.尚硅谷_SparkStreaming - DStream创建 - 自定义数据采集器】07:36

3.2 自定义数据源

3.2.1 用法及说明

3.2.2 案例实操

需求：自定义数据源，实现监控某个端口号，获取该端口号内容。

package com.atguigu.bigdata.spark.streamingimport java.util.Randomimport org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.ReceiverInputDStream
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}import scala.collection.mutableobject SparkStreaming03_DIY {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(3))val messageDS: ReceiverInputDStream[String] = ssc.receiverStream(new MyReceiver())messageDS.print()ssc.start()ssc.awaitTermination()}/*自定义数据采集器1.继承Receiver，定义泛型, 传递参数2.重写方法*/class MyReceiver extends Receiver[String](StorageLevel.MEMORY_ONLY) {private var flg = trueoverride def onStart(): Unit = {new Thread(new Runnable {override def run(): Unit = {while (flg) {val message = "采集的数据为：" + new Random().nextInt(10).toStringstore(message)Thread.sleep(500)}}}).start()}override def onStop(): Unit = {flg = false;}}
}

P191【191.尚硅谷_SparkStreaming - DStream创建 - Socket数据采集器源码解读】03:26

3.2.2 案例实操

需求：自定义数据源，实现监控某个端口号，获取该端口号内容。

P192【192.尚硅谷_SparkStreaming - DStream创建 - Kafka数据源】10:51

3.3 Kafka 数据源（面试、开发重点）

3.3.1 版本选型

3.3.2 Kafka 0-8 Receiver 模式（当前版本不适用）

3.3.3 Kafka 0-8 Direct 模式（当前版本不适用）

3.3.4 Kafka 0-10 Direct 模式

package com.atguigu.bigdata.spark.streamingimport java.util.Randomimport org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.{InputDStream, ReceiverInputDStream}
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.receiver.Receiver
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming04_Kafka {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(3))val kafkaPara: Map[String, Object] = Map[String, Object](ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG -> "linux1:9092,linux2:9092,linux3:9092",ConsumerConfig.GROUP_ID_CONFIG -> "atguigu","key.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer","value.deserializer" -> "org.apache.kafka.common.serialization.StringDeserializer")val kafkaDataDS: InputDStream[ConsumerRecord[String, String]] = KafkaUtils.createDirectStream[String, String](ssc,LocationStrategies.PreferConsistent,ConsumerStrategies.Subscribe[String, String](Set("atguiguNew"), kafkaPara))kafkaDataDS.map(_.value()).print()ssc.start()ssc.awaitTermination()}
}

第4章 DStream转换

P193【193.尚硅谷_SparkStreaming - DStream转换 - 状态操作】16:09

第 4 章 DStream 转换

DStream 上的操作与 RDD 的类似，分为 Transformations（转换）和 Output Operations（输出）两种，此外转换操作中还有一些比较特殊的原语，如：updateStateByKey()、transform()以及各种 Window 相关的原语。

4.1 无状态转化操作

package com.atguigu.bigdata.spark.streamingimport org.apache.kafka.clients.consumer.{ConsumerConfig, ConsumerRecord}
import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.InputDStream
import org.apache.spark.streaming.kafka010.{ConsumerStrategies, KafkaUtils, LocationStrategies}
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming05_State {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(3))ssc.checkpoint("cp")// 无状态数据操作，只对当前的采集周期内的数据进行处理// 在某些场合下，需要保留数据统计结果（状态），实现数据的汇总// 使用有状态操作时，需要设定检查点路径val datas = ssc.socketTextStream("localhost", 9999)val wordToOne = datas.map((_, 1))//val wordToCount = wordToOne.reduceByKey(_+_)// updateStateByKey：根据key对数据的状态进行更新// 传递的参数中含有两个值// 第一个值表示相同的key的value数据// 第二个值表示缓存区相同key的value数据val state = wordToOne.updateStateByKey((seq: Seq[Int], buff: Option[Int]) => {val newCount = buff.getOrElse(0) + seq.sumOption(newCount)})state.print()ssc.start()ssc.awaitTermination()}
}

P194【194.尚硅谷_SparkStreaming - DStream转换 - 无状态操作 - transform】09:06

4.1.1 Transform

package com.atguigu.bigdata.spark.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming06_State_Transform {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(3))val lines = ssc.socketTextStream("localhost", 9999)// transform方法可以将底层RDD获取到后进行操作// 1. DStream功能不完善// 2. 需要代码周期性地执行// Code : Driver端val newDS: DStream[String] = lines.transform(rdd => {// Code : Driver端，（周期性执行）rdd.map(str => {// Code : Executor端str})})// Code : Driver端val newDS1: DStream[String] = lines.map(data => {// Code : Executor端data})ssc.start()ssc.awaitTermination()}
}

P195【195.尚硅谷_SparkStreaming - DStream转换 - 无状态操作 - join】03:59

4.1.2 join

两个流之间的 join 需要两个流的批次大小一致，这样才能做到同时触发计算。计算过程就是对当前批次的两个流中各自的 RDD 进行 join，与两个 RDD 的 join 效果相同。

package com.atguigu.bigdata.spark.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming06_State_Join {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(5))val data9999 = ssc.socketTextStream("localhost", 9999)val data8888 = ssc.socketTextStream("localhost", 8888)val map9999: DStream[(String, Int)] = data9999.map((_, 9))val map8888: DStream[(String, Int)] = data8888.map((_, 8))// 所谓的DStream的Join操作，其实就是两个RDD的joinval joinDS: DStream[(String, (Int, Int))] = map9999.join(map8888)joinDS.print()ssc.start()ssc.awaitTermination()}
}

P196【196.尚硅谷_SparkStreaming - DStream转换 - 有状态操作 - window】12:17

4.2.2 WindowOperations

package com.atguigu.bigdata.spark.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}object SparkStreaming06_State_Window {def main(args: Array[String]): Unit = {val sparkConf = new SparkConf().setMaster("local[*]").setAppName("SparkStreaming")val ssc = new StreamingContext(sparkConf, Seconds(3))val lines = ssc.socketTextStream("localhost", 9999)val wordToOne = lines.map((_, 1))// 窗口的范围应该是采集周期的整数倍// 窗口可以滑动的，但是默认情况下，一个采集周期进行滑动// 这样的话，可能会出现重复数据的计算，为了避免这种情况，可以改变滑动的幅度（步长）val windowDS: DStream[(String, Int)] = wordToOne.window(Seconds(6), Seconds(6))val wordToCount = windowDS.reduceByKey(_ + _)wordToCount.print()ssc.start()ssc.awaitTermination()}
}