Spark大数据分析与实战（Python+PySpark）课件 第４章 SparkStreaming流数据计算

第４章SparkStreaming流数据计算Spark大数据分析与实战01SparkStreaming基本原理SparkStreaming词频统计0203Dstream数据转换操作目录CONTENTS04DStream输出操作05DStream数据源读取01SparkStreaming基本原理SparkStreaming基本原理SparkStreaming基本原理批计算一般是先有全量数据，然后将计算应用于这些数据，因此计算结果是一次性输出的。流计算的输入数据是时间上无界的持续，也永远拿不到完整的全量数据进行分析，且结果也是持续输出的，在时间上没有边界，就像水流一样，数据连绵不断地产生并被快速处理。流计算适用于有一定实时性要求的场合，为了提高计算效率，往往尽可能采用增量计算的方式，已计算过的数据不再重复处理SparkStreaming建立在SparkRDD基础之上，它接收实时输入的数据流，然后将数据拆分成多个批次，并将每个批次的数据交给Spark计算引擎处理，这样每个批次的数据就相当于一个局部范围内的离线数据SparkStreaming基本原理SparkStreaming设计了一个名为DStream的数据结构（离散化数据流），代表连续不断的数据流，实现上是将输入数据流按照预设的时间片进行分段，比如每秒切分一次，每个切分的数据段都被Spark转换成一个批次的RDD数据，这些数据分段则为DStream由于DStream是由切分数据段的RDD构成的，所以DStream的数据处理也就变成了对RDD的操作。在DStream数据集上应用的任何算子，比如map、reduce、join、window等，Spark底层都会将其翻译为对DStream的局部RDD的操作，并生成一个新的DStream02SparkStreaming词频统计Netcat网络工具测试DStream词频统计Netcat网络工具测试Ubuntu系统已经附带Netcat网络工具文件名为nc。我们可以开启两个终端窗体，其中左边终端窗体充当监听9999端口的服务端，右边终端窗体充当连接到服务器的客户端，然后双方互发数据并在对方显示如果一切正常，nc服务端和客户端就可以互发数据，每次输入的内容都会在对方的窗体显示，这是因为nc同时支持双向交互操作。测试完毕，按Ctrl+C快捷键结束右侧终端窗体中运行的nc客户端，保留左边的服务端窗体DStream词频统计SparkStreaming接收的是一个源源不断产生数据的数据源，是不间断循环运行的。在SparkStreaming应用程序中至少需要启动两个线程，其中一个用来接收数据，另一个用来处理数据，如果只有一个线程，就无法对数据进行处理，也就看不到任何实质性的效果DStream词频统计NetworkWordCount.py具体代码如下：DStream词频统计接下来通过spark-submit命令将SparkStreaming应用程序提交到Spark中运行当NetworkWordCount.py程序运行之后，每隔3秒就会处理并输出一次结果。接下来找到前面运行nc服务端的终端窗体，在其中随便输入一些单词，稍等片刻即可在SparkStreaming应用程序中打印词频统计的结果信息DStream词频统计Spark在运行时默认会产生大量日志信息，此时可在Spark的conf目录中通过perties配置文件设定全局的运行日志级别，或在代码中使用sc.setLogLevel()方法设定当前应用程序的运行日志级别，可以是WARN、ERROR、INFO、DEBUG、ALL等几种保存修改并退出编辑器，重新执行spark-submit命令将NetworkWordCount.py程序提交运行，并在nc服务端窗体再次输入一些单词，此时就可以比较清楚地查看SparkStreaming处理后的输出结果，避免了很多干扰信息03DStream转换操作DStream无状态转换操作DStream基于状态更新的转换DStream基于滑动窗口的转换DStream无状态转换操作DStream无状态转换操作仅仅计算当前时间片的数据内容，每个批次的处理结果不依赖于先前批次的数据，也不影响后续批次的数据。所以，DStream无状态转换操作处理的批次数据都是独立的，转换操作被直接应用到每个批次的RDD数据上，DStream的无状态转换操作与RDD的转换操作是类似的，返回的也是一个新的DStream对象除了无状态转换操作，DStream还支持有状态的转换操作，在这种情况下，DStream在计算当前批次数据时，会依赖之前批次的数据或中间结果，并不断把当前计算的数据与历史时间片的数据进行累计DStream基于状态更新的转换DStream基于状态更新的转换允许将当前时间和历史时间片的RDD数据叠加计算。比如，对DStream的数据按key执行reduce操作，然后将各个批次的中间结果累加到一起。Spark并非保存所有的历史数据，只是将当前的计算结果保存到磁盘以便下一次计算调用，避免重复计算DStream基于状态更新的转换在使用updateStateByKey算子时，必须开启checkpoint（检查点）机制并设置中间结果数据保存的目录（集群环境一般为HDFS上的目录，开发测试阶段可指定使用本地目录），这样才能把每个key对应的状态值长期保存，避免内存数据的丢失。比如，统计“双十一”当天的总销量和成交金额，操作者会在各个时段分批计算和局部累计产生的数据，最后进行全部汇总DStream基于状态更新的转换通过spark-submit命令将NetworkWordCountAll.py程序提交到Spark运行。需要注意的是，要确保nc服务端正在运行，否则需要重新执行nc-lk9999命令来启动它，不然在提交时会遇到错误根据输出结果可以看出，通过updateStateByKey算子统计的词频，都是在之前批次的中间统计结果基础上累积的数值，而不是从最开始的单词重复计算一遍后的数值DStream基于滑动窗口的转换DStream基于滑动窗口的转换是在时间轴上设置批次数据所在的“时间窗口”大小，以及窗口滑动的间隔，从而动态获取数据流的一种机制，比如每隔10秒统计一次最近30分钟的新闻热搜词。它是在一个比单批次间隔更长的时间范围内，通过整合位于窗口范围内的多个批次数据计算得到的数据结果window1=time1+time2+time3window3=window1+time4+time5-time1-time2window5=time3+time4+time5基于滑动窗口的转换是Spark提供的一组“短线”操作，它比单批次的“局部时间”长，但又比基于状态更新的“全程时间”短，通过滑动窗口技术可以实现大规模数据的增量更新和统计，即对任意一段时间内的数据进行处理，且不重复计算已处理过的数据DStream基于滑动窗口的转换SparkStreaming支持的滑动窗口算子包括reduceByKeyAndWindow、reduceByWindow、window等多种。下面以reduceByKeyAndWindow算子为例对词频统计例子进行改造，以实现基于滑动窗口的词频统计功能DStream基于滑动窗口的转换保存修改的代码并退出，确保nc服务端在监听9999端口，然后将修改后的代码提交到Spark运行，然后在nc服务端输入文字内容在nc服务端所在的终端窗体任意输入一些内容，此时SparkStreaming就会通过滑动窗口方式统计出相应的词频结果04DStream输出操作DStream输出操作DStream输出操作DStream可以根据需要输出到外部使用或保存，比如发送给Kafka消息系统，保存到数据库或者文件等。下面列出一些常用的DStream输出操作算子DStream输出操作以saveAsTextFiles算子为例，将词频统计结果保存到文本文件中DStream输出操作保存以上代码并退出编辑器，确保nc服务端在监听9999端口，然后在Linux终端窗体中通过spark-submit命令将NetworkWordCountSave.py程序提交到Spark运行，然后在nc服务端输入文字内容查看主目录中的streaming文件夹，里面出现了很多以output-开关的子目录，其中保存的就是不同批次时间点的数据文件05DStream数据源读取读取文件数据流读取Kafka数据流读取文件数据流SparkStreaming能够从本地文件目录、HDFS文件系统中读取数据，只需通过调用textFileStream()方法即可创建一个基于文件流类型的数据源读取文件数据流将代码提交到Spark运行再新打开一个Linux终端窗体，切换到~/streaming目录，使用echo命令依次在logfile子目录中创建1.txt和2.txt文件，同时观察运行代码的终端窗体所发生的变化SparkStreaming会持续监视logfile目录中的新文件，一旦处理后就不会再重复读取(即使修改文件内容也会被忽略)。如果是使用HDFS文件，只需将textFileStream()方法的路径参数修改为HDFS目录，比如hdfs://localhost:9000/logfile，然后不断地上传新文件到HDFS目录读取Kafka数据流（1）Kafka介绍Kafka是一个分布式的消息“发布-订阅”系统，也被称为消息中间件，它通过一个强大的消息队列处理大量的数据，并能够将消息从一个端点可靠地传递到另一个端点。Kafka非常适合离线和实时的数据消费，支持将消息内容保存在磁盘以防止数据丢失，能方便地与Spark集成，用于实时的流数据计算生产者和消费者（producer和consumer）：消息的发送者是producer，消息的使用和接收者是consumer。生产者将数据保存到Kafka中，消费者从Kafka中获取消息读取Kafka数据流（1）Kafka介绍Kafka实例（broker）：Kafka集群中有多个节点，每个节点都可以存储消息，每个节点就是一个Kafka实例，也被称为broker，其字面含义是“经纪人”主题（topic）：一个topic保存的是同一类消息，相当于消息分类。每个生产者将消息发送到Kafka时，必须指明要保存到哪个topic，以指明这个消息属于哪一类。topic的消息格式一般包含key和value两个字段，是一个（K,V）键值对形式的二元组分区（partition）：每个topic又可以分为多个partition，每个分区在磁盘上就是一个追加模式的log文件。任何发布到此partition的消息都会被追加到对应log文件的尾部。设置分区的原因是，Kafka基于文件进行存储，当文件内容多到一定程度时就很容易达到单个磁盘文件的上限，而采用一个分区对应一个文件的做法，数据可以被分别存储到不同的节点上，还能进行负载均衡读取Kafka数据流（1）Kafka介绍偏移量（offset）：一个分区是磁盘上的一个文件，消息存储在文件中的位置就称为offset，即偏移量，它用来唯一标记一条消息的整数。由于Kafka没有提供其他索引机制存储偏移量，因此文件只能按顺序操作，不允许随机读/写消息读取Kafka数据流（2）Kafka安装与测试启动Kafka依赖的ZooKeeper服务读取Kafka数据流（2）Kafka安装与测试将Kafka服务启动，启动完毕后，正常情况下会出现一个名为Kafka的进程名称至此，Kafka就运行起来了，Java进程中也多了一个名为Kafka的进程，默认使用9092端口读取Kafka数据流（2）Kafka安装与测试接下来运行程序先对Kafka进行简单的测试，我们在第1个终端窗体中创建mytopic主题，并向其发送几条测试的消息内容，然后在第2个终端窗体中获取mytopic主题收到的消息并显示将运行以上命令的Linux终端窗体称为A，代表消息的生产者读取Kafka数据流（2）Kafka安装与测试新打开一个Linux终端窗体B（代表消费者），在里面输入下面的命令准备完毕，现在测试Kafka的消息生产和消费。在终端窗体A中随便输