甘肃省住房和城乡建设厅注册中心网站,千博企业网站管理系统2013,知识营销案例,行业网站推广外包文章目录 1. 流式词频统计1.1 Spark Streaming编程步骤1.2 流式词频统计项目1.2.1 创建项目1.2.2 添加项目依赖1.2.3 修改源目录1.2.4 添加scala-sdk库1.2.5 创建日志属性文件 1.3 创建词频统计对象1.4 利用nc发送数据1.5 启动应用#xff0c;查看结果 2. 编程模型的基本概念3… 文章目录 1. 流式词频统计1.1 Spark Streaming编程步骤1.2 流式词频统计项目1.2.1 创建项目1.2.2 添加项目依赖1.2.3 修改源目录1.2.4 添加scala-sdk库1.2.5 创建日志属性文件 1.3 创建词频统计对象1.4 利用nc发送数据1.5 启动应用查看结果 2. 编程模型的基本概念3. 离散化数据流4. 基本数据源5. 基本DStream转换操作6. DStream输出操作 1. 流式词频统计
本实战演示了如何使用 Spark Streaming 实现实时词频统计。通过创建 Spark Streaming 项目添加依赖编写 Scala 代码监听网络端口接收数据流并按批次处理数据。利用 nc 工具发送数据程序每10秒统计一次词频并输出结果。该示例展示了 Spark Streaming 的微批处理特性适用于实时数据处理场景。
1.1 Spark Streaming编程步骤
添加SparkStreaming相关依赖获取程序入口接收数据对数据进行业务处理获取最终结果启动程序等待程序执行结束
1.2 流式词频统计项目
1.2.1 创建项目
设置项目基本信息 单击【Create】按钮生成项目基本骨架
1.2.2 添加项目依赖
在pom.xml文件里添加依赖
?xml version1.0 encodingUTF-8?
project xmlnshttp://maven.apache.org/POM/4.0.0xmlns:xsihttp://www.w3.org/2001/XMLSchema-instancexsi:schemaLocationhttp://maven.apache.org/POM/4.0.0http://maven.apache.org/xsd/maven-4.0.0.xsdmodelVersion4.0.0/modelVersiongroupIdnet.huawei.streaming/groupIdartifactIdSparkStreamingDemo/artifactIdversion1.0-SNAPSHOT/versionpropertiesmaven.compiler.source8/maven.compiler.sourcemaven.compiler.target8/maven.compiler.targetproject.build.sourceEncodingUTF-8/project.build.sourceEncoding/propertiesdependenciesdependencygroupIdorg.apache.spark/groupIdartifactIdspark-core_2.12/artifactIdversion3.3.0/version/dependencydependencygroupIdorg.apache.spark/groupIdartifactIdspark-sql_2.12/artifactIdversion3.3.0/version/dependencydependencygroupIdorg.apache.spark/groupIdartifactIdspark-streaming_2.12/artifactIdversion3.3.0/version/dependencydependencygroupIdjunit/groupIdartifactIdjunit/artifactIdversion4.13.2/version/dependency/dependencies/project刷新项目依赖
1.2.3 修改源目录 将java修改为scala 在pom.xml里设置源目录
1.2.4 添加scala-sdk库
在项目结构对话里添加 单击【Add to Modules】菜单项 单击【OK】按钮以后就可以在scala里创建Scala Class了
1.2.5 创建日志属性文件
在resources里创建log4j2.properties文件
rootLogger.level ERROR
rootLogger.appenderRef.stdout.ref consoleappender.console.type Console
appender.console.name console
appender.console.layout.type PatternLayout
appender.console.layout.pattern %d{yy/MM/dd HH:mm:ss} %p %c{1}: %m%n%ex1.3 创建词频统计对象
创建net.huawei.streaming包 在net.huawei.streaming包里创建SparkStreamingWordCount对象
package net.huawei.streamingimport org.apache.spark.SparkConf
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}
import org.apache.spark.streaming.{Seconds, StreamingContext}/*** 功能流式词频统计* 作者华卫* 日期2025年01月23日*/
object SparkStreamingWordCount {def main(args: Array[String]): Unit {// 创建SparkConf对象2个线程本地运行val conf: SparkConf new SparkConf().setMaster(local[2]).setAppName(SparkStreamingWordCount)// 创建StreamingContext对象10秒一个批次val ssc: StreamingContext new StreamingContext(conf, Seconds(10))// 创建ReceiverInputDStream对象接收来自网络端口的数据val lines: ReceiverInputDStream[String] ssc.socketTextStream(bigdata1, 9999)// lines中每条数据按照空格进行切分然后扁平化处理val words: DStream[String] lines.flatMap(_.split( ))// words中每条数据转换成(word,1)二元组val wordmap: DStream[(String, Int)] words.map(word (word, 1))// wordmap中每条数据按key分组按value进行累加求和val wordcount: DStream[(String, Int)] wordmap.reduceByKey(_ _)// 打印词频统计结果 wordcount.print()// 启动实时流程序ssc.start()// 等待实时流程序结束ssc.awaitTermination()}
}代码说明这段代码实现了一个基于Spark Streaming的实时词频统计程序。它通过监听指定端口bigdata1:9999接收数据流将每行数据按空格切分并扁平化为单词然后统计每个单词的出现次数。程序每10秒处理一个批次的数据并打印词频统计结果。代码结构清晰适用于实时数据处理场景。
1.4 利用nc发送数据
在bigdata1节点利用nc发送数据执行命令nc -lp 9999
1.5 启动应用查看结果
启动SparkStreamingWordCount对象在bigdata1节点上输入数据在控制台查看词频统计结果 结果说明Spark Streaming 采用微批处理每批次数据独立处理批次间不共享状态或共同计数。默认情况下批次间数据互不影响。如需跨批次状态管理可使用 updateStateByKey 或 mapWithState 实现累加计数等功能。这种设计确保了流数据处理的灵活性和高效性。
2. 编程模型的基本概念
3. 离散化数据流
4. 基本数据源
5. 基本DStream转换操作
6. DStream输出操作
