【Spark分布式内存计算框架——Spark Streaming】3.入门案例（上）官方案例运行

news/2024/4/26 17:34:03/文章来源:https://blog.csdn.net/CSDNGuoYuying/article/details/129220190

2.1 官方案例运行

运行官方提供案例，使用【$SPARK_HOME/bin/run-example】命令运行，效果如下：
在这里插入图片描述

具体步骤如下：

第一步、准备数据源启动端口，准备数据

nc -lk 9999
spark spark hive hadoop spark hive

第二步、运行官方案例

使用官方提供命令行运行案例

# 官方入门案例运行：词频统计
/export/server/spark/bin/run-example --master local[2] streaming.NetworkWordCount node1.itcast.cn 9999

第三步、运行结果
在这里插入图片描述
SparkStreaming模块对流式数据处理，介于Batch批处理和RealTime实时处理之间处理数据方式。

2.2 编程实现

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

StreamingContext
回顾SparkCore和SparkSQL及SparkStreaming处理数据时编程：

1）、SparkCore

数据结构：RDD
SparkContext：上下文实例对象

2）、SparkSQL

数据结构：Dataset/DataFrame = RDD + Schema
SparkSession：会话实例对象，在Spark 1.x中SQLContext/HiveContext

3）、SparkStreaming

数据结构：DStream = Seq[RDD]
StreamingContext：流式上下文实例对象，底层还是SparkContext
参数：划分流式数据时间间隔BatchInterval：1s，5s（演示）

文档：http://spark.apache.org/docs/2.2.0/streaming-programming-guide.html#initializing-streamingcontext
从官方文档可知，提供两种方式构建StreamingContext实例对象，截图如下：

第一种方式：构建SparkConf对象
在这里插入图片描述
第二种方式：构建SparkContext对象

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

1、Define the input sources by creating input DStreams.
定义从哪个数据源接收流式数据，封装到DStream中
2、Define the streaming computations by applying transformation and output operations to DStreams.
针对业务调用DStream中函数，进行数据处理和输出
3、Start receiving data and processing it using streamingContext.start().
4 、 Wait for the processing to be stopped (manually or due to any error) using streamingContext.awaitTermination().
5、The processing can be manually stopped using streamingContext.stop().
启动流式应用，并且一直等待程序终止（人为或异常），最后停止运行

完整StreamingWordCount代码如下所示：

import org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}
/**
* 基于IDEA集成开发环境，编程实现从TCP Socket实时读取流式数据，对每批次中数据进行词频统计。
*/
object StreamingWordCount {
def main(args: Array[String]): Unit = {
// TODO: 1. 构建StreamingContext流式上下文实例对象
val ssc: StreamingContext = {
// a. 创建SparkConf对象，设置应用配置信息
val sparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[3]")
// b.创建流式上下文对象, 传递SparkConf对象，TODO: 时间间隔 -> 用于划分流式数据为很多批次Batch
val context = new StreamingContext(sparkConf, Seconds(5))
// c. 返回
context
}
// TODO: 2. 从数据源端读取数据，此处是TCP Socket读取数据
/*
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. 对每批次的数据进行词频统计
val resultDStream: DStream[(String, Int)] = inputDStream
// 过滤不合格的数据
.filter(line => null != line && line.trim.length > 0)
// 按照分隔符划分单词
.flatMap(line => line.trim.split("\\s+"))
// 转换数据为二元组，表示每个单词出现一次
.map(word => (word, 1))
// 按照单词分组，聚合统计
.reduceByKey((tmp, item) => tmp + item)
// TODO: 4. 将结果数据输出 -> 将每批次的数据处理以后输出
resultDStream.print(10)
// TODO: 5. 对于流式应用来说，需要启动应用
ssc.start()
// 流式应用启动以后，正常情况一直运行（接收数据、处理数据和输出数据），除非人为终止程序或者程序异常停止
ssc.awaitTermination()
// 关闭流式应用(参数一：是否关闭SparkContext，参数二：是否优雅的关闭）
ssc.stop(stopSparkContext = true, stopGracefully = true)
}
}