Spark StreamingSpark Day10（Spark Streaming 学习笔记） _streaming

将相本无种，男儿当自强。这篇文章主要讲述Spark StreamingSpark Day10：Spark Streaming 学习笔记相关的知识，希望能为你提供帮助。

Spark Day10：Spark Streaming

文章图片

01-[了解]-昨日课程内容回顾

实战练习：以DMP广告行业背景为例，处理广告点击数据，分为2个方面【广告数据ETL转换和业务报表开发】，具体说明如下：

【前提】：使用SparkSQL完成案例练习，进行代码编写
1、广告数据ETL转换
JSON文本数据 -> DataFrame：提取IP地址，解析转换为省份和城市 -> 保存到Hive分区表中
数据源
文件系统（HDFS、LocalFS）文本文件数据：JSON格式
数据处理
ip地址，转换省份与城市
实现：使用DSL编程，可以调用类似SQL语句函数、也可以调用类似RDD转换函数，比如mapPartitions
数据终端Sink
Hive分区表

2、业务报表分析
【前提】：默认情况下，每次分析前一天数据
数据流程：
Hive分区表-> DataFrame：依据业务进行报表分析统计 -> mysql数据库表中
【注意】：
a. 加载数据时，考虑过滤，仅仅获取前一天数据
b. 报表分析时
使用SQL编程，比较容易
可以考虑DSL编程
c. 保存数据时
不能直接使用SparkSQL提供外部数据源接口，使用原生态JDBC
dataframe.rdd.foreachPartition(iter => saveToMySQL(iter))

【扩展】：向MySQL表写入数据时，实现upsert【主键存在，就更新；不存在，就插入】功能
- 方式一：使用REPALCE 代替INSERT
replace INTO db_test.tb_wordcount (word, count) VALUES(?, ?)
此种方式，有一些限制，比如需要列举出所有列等

- 方式二：ON DUPLICATE KEY UPDATE
INSERT INTO ods_qq_group_members ( gid, uin, datadate )
VALUES (111, 1111111, 2016-11-29 )
ON DUPLICATE KEY UPDATE gid=222, uin=22222, datadate=2016-11-29

02-[了解]-今日课程内容提纲

从今天开始，进入Spark框架中：关于流式数据分析模块讲解。

SparkCore与SparkSQL，离线分析批处理，分析数据都是静态的，不变的
SparkStreaming和StructuredStreaming，实时流式数据分析，分析数据是源源不断产生，一产生就进行分析
首先，学习SparkStreaming流式计算模块，以批处理思想处理流式数据，进行实时分析。

1、Streaming 流式计算概述
Streaming应用场景，目前需求非常多
Lambda 架构，离线和实时
Streaming 计算模式
SparkStreaming 计算思想

2、入门案例
官方案例运行，”词频统计“
编程实现代码：SparkStreaming入门程序编写
Streaming 工作原理
如何使用批的思想处理流式数据

3、DStream：分离、离散流
DStream是什么，DStream = Seq[RDD]
DStream Operations
函数，分为2类：转换函数、输出函数
流式应用状态

文章图片

03-[了解]-Spark框架中各个模块的数据结构抽象

Spark框架是一个统一分析引擎，包含很多模块，各个模块都有数据结构封装数据。

文章图片

在Spark1.x时，主要三个模块，都是自己数据结构进行封装
- SparkCore：RDD
- SparkSQL：DataFrame/Dataset
- SparkStreaming：DStream

到Spark2.x时，建议使用SparkSQL对离线数据和流式数据分析
Dataset/DataFrame
出现StructuredStreaming模块，将流式数据封装到Dataset中，使用DSL和SQL分析流式数据

04-[了解]-Straming 概述之流式应用场景

1）、电商实时大屏：每年双十一时，淘宝和京东实时订单销售额和产品数量大屏展示

文章图片

2）、商品推荐：京东和淘宝的商城在购物车、商品详情等地方都有商品推荐的模块

文章图片

3）、工业大数据：现在的工场中, 设备是可以联网的, 汇报自己的运行状态, 在应用层可以针对
这些数据来分析运行状况和稳健程度, 展示工件完成情况, 运行情况等

文章图片

4）、集群监控：一般的大型集群和平台, 都需要对其进行监控

文章图片

具体来说，流式计算应用场景如下所示：

文章图片

05-[掌握]-Straming 概述之Lambda架构

Lambda架构是由Storm的作者Nathan Marz提出的一个实时大数据处理框架。

大数据架构，仅能够进行离线数据分析，又能够进行实时数据计算。

文章图片

Lambda架构分为三层：
- 第一层：Batch Layer
批处理层
数据离线分析
- 第二层：Speed Layer
速度层
数据实时计算
- 第三层：ServingLayer
服务层
将离线分析和实时计算结果对外提供服务，比如可视化展示

Lambda架构整合离线计算和实时计算，融合不可变性（Immunability），读写分离和复杂性隔离等一系列架构原则，可集成Hadoop，Kafka，Storm，Spark，Hbase等各类大数据组件。

文章图片

举个例子：
【每年双11狂欢购物节，当天用户交易订单数数据来说】
- 第一点：实时统计交易订单中销售额
totalAmt
最终实时在大屏展示
- 第二点：11.11以后，11.12凌晨，开始对前一天交易数据进行分析
哪个省份娘们最败家
哪个城市女性消费最优秀
无论实时计算还是离线分析，最终都需要展示
提供计算分析的数据

Lambda架构通过分解的三层架构来解决该问题：批处理层（Batch Layer），速度层（SpeedLayer）和服务层（Serving Layer）。

文章图片

06-[掌握]-Straming 概述之流式数据计算模式

目前大数据框架领域有如下几种流式计算框架：

1）、Storm框架
阿里巴巴双11，前几年使用就是此框架
2）、Samza，领英公司开源
严重依赖Kafka，在国内几乎没有公司使用
3）、SparkStreaming
基于SparkCore之上流式计算框架，目前使用也不多
4）、Flink 框架
当前大数据流式计算领域最流行框架，尤其在国内，推广十分广泛，各个大厂都在使用，实时性很高，吞吐量比较大，尤其在阿里巴巴公司。
5）、StructuredStreaming

SparkSQL框架中针对流式数据处理功能模块
从Spark2.0提出来，相对来说，比较优秀，很多公司在使用SparkSQL时，如果有流式数据需要实时处理的话，直接选择StructuredStreaming

不同的流式处理框架有不同的特点，也适应不同的场景，主要有如下两种模式。

总的来说，流式计算引擎（框架）处理流式数据有2中模式）

模式一：原生流处理（Native）

所有输入记录会一条接一条地被处理，上面提到的 Storm 和 Flink都是采用这种方式；

文章图片

产生一条数据，处理一条数据，此类框架处理数据速度非常快的，实时性很高

模式二：微批处理（Batch）

将输入的数据以某一时间间隔 T，切分成多个微批量数据，然后对每个批量数据进行处理，Spark Streaming 和 StructuredStreaming采用的是这种方式

文章图片

微批处理，将流式数据划分很多批次，往往按照时间间隔划分，比如1秒钟，进行处理分析

目前大数据框架领域有如下几种流式计算框架：

1）、Storm框架
阿里巴巴双11，前几年使用就是此框架
2）、Samza，领英公司开源
严重依赖Kafka，在国内几乎没有公司使用
3）、SparkStreaming
基于SparkCore之上流式计算框架，目前使用也不多
4）、Flink 框架
当前大数据流式计算领域最流行框架，尤其在国内，推广十分广泛，各个大厂都在使用，实时性很高，吞吐量比较大，尤其在阿里巴巴公司。
5）、StructuredStreaming
SparkSQL框架中针对流式数据处理功能模块
从Spark2.0提出来，相对来说，比较优秀，很多公司在使用SparkSQL时，如果有流式数据需要实时处理的话，直接选择StructuredStreaming
—

07-[掌握]-Straming 概述之SparkStreaming计算思想

? Spark Streaming是Spark生态系统当中一个重要的框架，它??建立在Spark Core之上??，下图也可以看出Sparking Streaming在Spark生态系统中地位。

文章图片

官方定义Spark Streaming模块：

文章图片

SparkStreaming使用户构建可扩展的、具有容错语义流式应用更加容易。

SparkStreaming是一个基于SparkCore之上的实时计算框架，可以从很多数据源消费数据并对数据进行实时的处理，具有高吞吐量和容错能力强等特点。

文章图片

对于Spark Streaming来说，将流式数据按照时间间隔BatchInterval划分为很多部分，每一部分Batch（批次），针对每批次数据Batch当做RDD进行快速分析和处理。- 第一点、按照时间间隔划分流式数据batchInterval，比如1秒- 第二点、划分数据当做批次Batch每批次数据认为是RDD- 第三点，处理流式数据时，仅仅处理每批次RDD即可RDD数据分析处理

文章图片

对于Spark中StructuredStreaming结构化六来说
- 默认情况下，属于微批处理模式
一批次一批次处理数据
- Spark 2.3开始，Continues Processing
持续流处理，就是原生流模式分析数据

文章图片

? ??将流式数据按照【X seconds】划分很多批次Batch，每个Batch数据封装到RDD中进行处理分析，最后每批次数据进行输出。??

文章图片

? 对于目前版本的Spark Streaming而言，其最小的??Batch Size的选取在0.5~5秒钟之间??，所以Spark Streaming能够满足流式准实时计算场景，

08-[掌握]-入门案例之运行官方词频统计

SparkStreaming官方提供Example案例，功能描述：??从TCP Socket数据源实时消费数据，对每批次Batch数据进行词频统计WordCount??，流程图如下：

文章图片

1、数据源：TCP Socket
从哪里读取实时数据，然后进行实时分析

2、数据终端：输出控制台
结果数据输出到哪里

3、功能：对每批次数据实时统计，时间间隔BatchInterval：1s

运行官方提供案例，使用【??$SPARK_HOME/bin/run-example??】命令运行，效果如下：

文章图片

具体步骤如下：

文章图片

SparkStreaming模块对流式数据处理，介于Batch批处理和RealTime实时处理之间处理数据方式。

文章图片

09-[掌握]-入门案例之Streaming编程模块

基于IDEA集成开发环境，编程实现：从TCP Socket实时读取流式数据，对每批次中数据进行词频统计WordCount。

在Spark框架中各个模块都有自己数据结构，也有自己的程序入口：
- SparkCore
RDD
SparkContext
- SparkSQL
DataFrame/Dataset
SparkSession/SQLContext(Spark 1.x)
- SparkStreaming
DStream
StreamingContext
参数：划分流式数据时间间隔BatchInterval：1s，5s（演示）
底层还是SparkContext，每批次数据当做RDD

从官方文档可知，提供两种方式构建StreamingContext实例对象，截图如下：

第一种方式：构建??SparkConf??对象

文章图片

第二种方式：构建SparkContext对象

文章图片

针对SparkStreaming流式应用来说，代码逻辑大致如下五个步骤：

文章图片

编写SparkStreaming程序模块，构建StreamingContext流式上下文实例对象，启动流式应用等待终止

package cn.itcast.spark.start

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object _01StreamingWordCount {

def main(args: Array[String]): Unit = {

// TODO: 1. 构建StreamingContext实例对象，传递时间间隔BatchInterval
val ssc: StreamingContext = {
// 创建SparkConf对象，设置应用属性
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// 传递sparkConf对象和时间间隔batchInterval：5秒
new StreamingContext(sparkConf, Seconds(5))
}

// TODO: 2. 定义数据源，获取流式数据，封装到DStream中

// TODO: 3. 依据业务需求，调用DStream中转换函数（类似RDD中转换函数）

// TODO: 4. 定义数据终端，将每批次结果数据进行输出

// TODO: 5. 启动流式应用，等待终止
ssc.start() // 启动流式应用，开始从数据源实时消费数据，处理数据和输出结果
// 流式应用只要已启动，一直运行，除非程序异常终止或者认为终止
ssc.awaitTermination()
// 当流式应用停止时，需要关闭资源
ssc.stop(stopSparkContext = true, stopGracefully = true)
}

}

10-[掌握]-入门案例之代码实现及测试运行

每个流式应用程序（无论是SparkStreaming、StructuredStreaming还是Flink），最核心有3个步骤

- 第一步、数据源Source
从哪里实时消费流式数据
- 第二步、数据转换Transformation
按照业务处理数据
调用函数
- 第三步、数据终端Sink
将处理结果数据保存到外部系统中

package cn.itcast.spark.start

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object _01StreamingWordCount {

def main(args: Array[String]): Unit = {

// TODO: 1. 构建StreamingContext实例对象，传递时间间隔BatchInterval
val ssc: StreamingContext = {
// 创建SparkConf对象，设置应用属性
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// 传递sparkConf对象和时间间隔batchInterval：5秒
new StreamingContext(sparkConf, Seconds(5))
}

// TODO: 2. 定义数据源，获取流式数据，封装到DStream中
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: ReceiverInputDStream[String] = ssc.socketTextStream("node1.itcast.cn", 9999)

// TODO: 3. 依据业务需求，调用DStream中转换函数（类似RDD中转换函数）
/*
spark hive hive spark spark hadoop
*/
val resultDStream: DStream[(String, Int)] = inputDStream
// 分割单词
.flatMap(line => line.trim.split("\\\\s+"))
// 转换为二元组
.map(word => word -> 1)
// 按照单词分组，组内进行聚合
/*
(spark, 1)
(spark, 1)-> (spark, [1, 1])(hive, [1]) -> (spark, 2) (hive, 1)
(hive, 1)
*/
.reduceByKey((tmp, itme) => tmp + itme)

// TODO: 4. 定义数据终端，将每批次结果数据进行输出
resultDStream.print()

// TODO: 5. 启动流式应用，等待终止
ssc.start() // 启动流式应用，开始从数据源实时消费数据，处理数据和输出结果
// 流式应用只要已启动，一直运行，除非程序异常终止或者认为终止
ssc.awaitTermination()
// 当流式应用停止时，需要关闭资源
ssc.stop(stopSparkContext = true, stopGracefully = true)
}

}

运行结果监控截图：

文章图片

11-[掌握]-入门案例之SparkStreaming 运行工作原理

SparkStreaming处理流式数据时，按照时间间隔划分数据为微批次（Micro-Batch），每批次数据当做RDD，再进行处理分析。

文章图片

以上述词频统计WordCount程序为例，讲解Streaming工作原理。

第一步、创建 StreamingContext

当SparkStreaming流式应用启动（streamingContext.start）时，首先创建StreamingContext流式上下文实例对象，整个流式应用环境构建，底层还是SparkContext。

文章图片

从WEB UI界面【Jobs Tab】可以看到【Job-0】是一个Receiver接收器，一直在运行，以Task方式运行，需要1Core CPU。

文章图片

第二步、接收器接收数据

? 启动每个接收器Receiver以后，实时从数据源端接收数据（比如TCP Socket），也是??按照时间间隔将接收的流式数据划分为很多Block（块）??。

文章图片

接收器 Receiver 划分流式数据的时间间隔 BlockInterval ，默认值为 200ms ，通过属性【spark.streaming.blockInterval】设置。

假设设置Batch批次时间间隔为1s，每批次默认情况下，有几个Block呢？？？
1s = 1000ms = 200ms * 5
所以5个block
将该批次数据当做1个RDD，此时RDD的分区数目为5
——

第3步、汇报接收Block报告

接收器Receiver将实时汇报接收的数据对应的Block信息，当BatchInterval时间达到以后，StreamingContext将对应时间范围内数据block当做RDD，加载SparkContext处理数据。

文章图片

以此循环处理流式的数据，如下图所示：

文章图片

12-[掌握]-DStream 是什么

SparkStreaming模块将流式数据封装的数据结构：??DStream??（Discretized Stream，离散化数据流，连续不断的数据流），代表持续性的数据流和经过各种Spark算子操作后的结果数据流。

文章图片

通过WEB UI界面可知，对DStream调用函数操作，底层就是对RDD进行操作，发现狠多时候DStream中函数与RDD中函数一样的。

文章图片

13-[了解]-DStream Operations函数概述

DStream类似RDD，里面包含很多函数，进行数据处理和输出操作，主要分为两大类：

文章图片

其一：转换函数【Transformation函数】

文章图片

在SparkStreaming中对流的转换操作，主要3种转换类型： - 对流中数据进行转换map、flatMpa、filter- 对流中数据涉及到聚合统计countreducecountByValue... - 对2个流进行聚合啊哦做unionjoincogroup

其二：输出函数【Output函数】

文章图片

? DStream中每批次结果RDD输出使用??foreachRDD???函数，前面使用的??print???函数底层也是调用??foreachRDD??函数，截图如下所示：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传

在DStream中有两个重要的函数，都是针对每批次数据RDD进行操作的，更加接近底层，性能更好，强烈推荐使用：

文章图片

14-[掌握]-DStream中transform函数使用

通过源码认识??transform??函数，有两个方法重载，声明如下：

文章图片

接下来使用transform函数，修改词频统计程序，具体代码如下：

package cn.itcast.spark.rdd

import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object _03StreamingTransformRDD {

def main(args: Array[String]): Unit = {

// TODO: 1. 构建StreamingContext实例对象，传递时间间隔BatchInterval
val ssc: StreamingContext = {
// a. 创建SparkConf对象，设置应用基本信息
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// b. 创建实例对象，设置BatchInterval
new StreamingContext(sparkConf, Seconds(5))
}

// TODO: 2. 定义数据源，获取流式数据，封装到DStream中
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: DStream[String] = ssc.socketTextStream(
"node1.itcast.cn",
9999,
storageLevel = StorageLevel.MEMORY_AND_DISK
)

// TODO: 3. 依据业务需求，调用DStream中转换函数（类似RDD中转换函数）
/*
TODO: 能对RDD操作的就不要对DStream操作，当调用DStream中某个函数在RDD中也存在，使用针对RDD操作
def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
*/
// 此处rdd就是DStream中每批次RDD数据
val resultDStream: DStream[(String, Int)] = inputDStream.transform{rdd =>
val resultRDD: RDD[(String, Int)] = rdd
// 按照分隔符分割单词
.flatMap(line => line.split("\\\\s+"))
// 转换单词为二元组，表示每个单词出现一次
.map(word => word -> 1)
// 按照单词分组，对组内执进行聚合reduce操作，求和
.reduceByKey((tmp, item) => tmp + item)
// 每批次RDD处理结果RDD返回
resultRDD
}

// TODO: 4. 定义数据终端，将每批次结果数据进行输出
resultDStream.print()

// TODO: 5. 启动流式应用，等待终止
ssc.start()
ssc.awaitTermination()
ssc.stop(stopSparkContext = true, stopGracefully = true)
}

}

查看WEB UI监控中每批次Batch数据执行Job的DAG图，直接显示针对RDD进行操作。

文章图片

15-[掌握]-DStream中foreachRDD函数使用

??foreachRDD???函数属于将DStream中结果数据RDD输出的操作，类似??transform??函数，针对每批次RDD数据操作，源码声明如下：

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传
继续修改词频统计代码，自定义输出数据，具体代码如下：

package cn.itcast.spark.output

import org.apache.commons.lang3.time.FastDateFormat
import org.apache.spark.SparkConf
import org.apache.spark.rdd.RDD
import org.apache.spark.storage.StorageLevel
import org.apache.spark.streaming.dstream.DStream
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object _04StreamingOutputRDD {

def main(args: Array[String]): Unit = {

// TODO: 1. 构建StreamingContext实例对象，传递时间间隔BatchInterval
val ssc: StreamingContext = {
// a. 创建SparkConf对象，设置应用基本信息
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// TODO：设置数据输出文件系统的算法版本为2
.set("spark.hadoop.mapreduce.fileoutputcommitter.algorithm.version", "2")
// b. 创建实例对象，设置BatchInterval
new StreamingContext(sparkConf, Seconds(5))
}

// TODO: 2. 定义数据源，获取流式数据，封装到DStream中
/*
def socketTextStream(
hostname: String,
port: Int,
storageLevel: StorageLevel = StorageLevel.MEMORY_AND_DISK_SER_2
): ReceiverInputDStream[String]
*/
val inputDStream: DStream[String] = ssc.socketTextStream(
"node1.itcast.cn",
9999,
storageLevel = StorageLevel.MEMORY_AND_DISK
)

// TODO: 3. 依据业务需求，调用DStream中转换函数（类似RDD中转换函数）
/*
TODO: 能对RDD操作的就不要对DStream操作，当调用DStream中某个函数在RDD中也存在，使用针对RDD操作
def transform[U: ClassTag](transformFunc: RDD[T] => RDD[U]): DStream[U]
*/
// 此处rdd就是DStream中每批次RDD数据
val resultDStream: DStream[(String, Int)] = inputDStream.transform{ rdd =>
val resultRDD: RDD[(String, Int)] = rdd
.filter(line => null != line & & line.trim.length > 0)
.flatMap(line => line.trim.split("\\\\s+"))
.map(word => (word, 1))
.reduceByKey((tmp, item) => tmp + item)
// 返回结果RDD
resultRDD
}

// TODO: 4. 定义数据终端，将每批次结果数据进行输出
//resultDStream.print()
/*
def foreachRDD(foreachFunc: (RDD[T], Time) => Unit): Unit
rdd 表示每批次处理结果RDD
time 表示批次产生的时间，Long类型
*/
resultDStream.foreachRDD((rdd, time) => {
// 打印每批次产生的时间
val batchTime: String = FastDateFormat.getInstance("yyyy/MM/dd HH:mm:ss").format(time.milliseconds)
println("-------------------------------------------")
println(s"Batch Time: ${batchTime}")
println("-------------------------------------------")

// TODO：判断结果RDD是否有数据，有数据，再进行输出，否则不操作
if(!rdd.isEmpty()){
// 对结果RDD进行输出时：降低分区数目、针对每个分区操作、通过连接池（sparkStreaming)获取连接
val resultRDD: RDD[(String, Int)] = rdd.coalesce(1)
resultRDD.cache()
// 将结果RDD 打印控制台
resultRDD.foreachPartition(iter => iter.foreach(println))

// 将结果RDD 保存文件中
resultRDD.saveAsTextFile(s"datas/streaming-wc-${time.milliseconds}")
resultRDD.unpersist()
}
})

// TODO: 5. 启动流式应用，等待终止
ssc.start()
ssc.awaitTermination()
ssc.stop(stopSparkContext = true, stopGracefully = true)
}

}

16-[了解]-SparkStreaming流式应用三种状态

使用SparkStreaming处理实际实时应用业务时，针对不同业务需求，需要使用不同的函数。SparkStreaming流式计算框架，针对具体业务主要分为三类，使用不同函数进行处理：

业务一：??无状态Stateless??

文章图片

业务二：??有状态State??

文章图片

业务三：??窗口统计??

[外链图片转存失败,源站可能有防盗链机制,建议将图片保存下来直接上传(img-g0Xbre1z-1638081078231)(/img/image-20210429115034287.png)]
附录一、创建Maven模块 1）、Maven 工程结构

文章图片

2）、POM 文件内容
Maven 工程POM文件中内容（依赖包）：

< !-- 指定仓库位置，依次为aliyun、cloudera和jboss仓库 -->
< repositories>
< repository>
< id> aliyun< /id>
< url> http://maven.aliyun.com/nexus/content/groups/public/< /url>
< /repository>
< repository>
< id> cloudera< /id>
< url> https://repository.cloudera.com/artifactory/cloudera-repos/< /url>
< /repository>
< repository>
< id> jboss< /id>
< url> http://repository.jboss.com/nexus/content/groups/public< /url>
< /repository>
< /repositories>

< properties>
< scala.version> 2.11.12< /scala.version>
< scala.binary.version> 2.11< /scala.binary.version>
< spark.version> 2.4.5< /spark.version>
< hadoop.version> 2.6.0-cdh5.16.2< /hadoop.version>
< hbase.version> 1.2.0-cdh5.16.2< /hbase.version>
< kafka.version> 2.0.0< /kafka.version>
< mysql.version> 8.0.19< /mysql.version>
< /properties>

< dependencies>

< !-- 依赖Scala语言 -->
< dependency>
< groupId> org.scala-lang< /groupId>
< artifactId> scala-library< /artifactId>
< version> ${scala.version}< /version>
< /dependency>
< !-- Spark Core 依赖 -->
< dependency>
< groupId> org.apache.spark< /groupId>
< artifactId> spark-core_${scala.binary.version}< /artifactId>
< version> ${spark.version}< /version>
< /dependency>
< !-- Spark SQL 依赖 -->
< dependency>
< groupId> org.apache.spark< /groupId>
< artifactId> spark-sql_${scala.binary.version}< /artifactId>
< version> ${spark.version}< /version>
< /dependency>
< !-- Spark Streaming 依赖 -->
< dependency>
< groupId> org.apache.spark< /groupId>
< artifactId> spark-streaming_${scala.binary.version}< /artifactId>
< version> ${spark.version}< /version>
< /dependency>
< !-- Spark Streaming 集成Kafka 0.8.2.1 -->
< dependency>
< groupId> org.apache.spark< /groupId>
< artifactId> spark-streaming-kafka-0-8_${scala.binary.version}< /artifactId>
< version> ${spark.version}< /version>
< /dependency>
< !-- Spark Streaming 与Kafka 0.10.0 集成依赖-->
< dependency>
< groupId> org.apache.spark< /groupId>
< artifactId> spark-streaming-kafka-0-10_${scala.binary.version}< /artifactId>
< version> ${spark.version}< /version>
< /dependency>
< !-- Hadoop Client 依赖 -->
< dependency>
< groupId> org.apache.hadoop< /groupId>
< artifactId> hadoop-client< /artifactId>
< version> ${hadoop.version}< /version>
< /dependency>
< !-- HBase Client 依赖 -->
< dependency>
< groupId> org.apache.hbase< /groupId>
< artifactId> hbase-server< /artifactId>
< version> ${hbase.version}< /version>
< /dependency>
< dependency>
< groupId> org.apache.hbase< /groupId>
< artifactId> hbase-hadoop2-compat< /artifactId>
< version> ${hbase.version}< /version>
< /dependency>
< dependency>
< groupId> org.apache.hbase< /groupId>
< artifactId> hbase-client< /artifactId>
< version> ${hbase.version}< /version>
< /dependency>
< !-- Kafka Client 依赖 -->
< dependency>
< groupId> org.apache.kafka< /groupId>
< artifactId> kafka-clients< /artifactId>
< version> 2.0.0< /version>
< /dependency>
< !-- 根据ip转换为省市区 -->
< dependency>
< groupId> org.lionsoul< /groupId>
< artifactId> ip2region< /artifactId>
< version> 1.7.2< /version>
< /dependency>
< !-- MySQL Client 依赖 -->
< dependency>
< groupId> mysql< /groupId>
< artifactId> mysql-connector-java< /artifactId>
< version> ${mysql.version}< /version>
< /dependency>
< dependency>
< groupId> c3p0< /groupId>
< artifactId> c3p0< /artifactId>
< version> 0.9.1.2< /version>
< /dependency>
< /dependencies>

< build>
< outputDirectory> target/classes< /outputDirectory>
< testOutputDirectory> target/test-classes< /testOutputDirectory>
< resources>
< resource>
< directory> ${project.basedir}/src/main/resources< /directory>
< /resource>
< /resources>
< !-- Maven 编译的插件 -->
< plugins>
< plugin>
< groupId> org.apache.maven.plugins< /groupId>
< artifactId> maven-compiler-plugin< /artifactId>
< version> 3.0< /version>
< configuration>
< source> 1.8< /source>
< target> 1.8< /target>
< encoding> UTF-8< /encoding>
< /configuration>
< /plugin>
< plugin>
< groupId> net.alchim31.maven< /groupId>
< artifactId> scala-maven-plugin< /artifactId>
< version> 3.2.0< /version>
< executions>
< execution>
< goals>
< goal> compile< /goal>
< goal> testCompile< /goal>
< /goals>
< /execution>
< /executions>
< /plugin>
< /plugins>
< /build>