当前位置：首页 > news >正文

Spark Streaming编程基础

news 来源：原创 2025/5/26 13:00:50

文章目录

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 version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>net.huawei.streaming</groupId><artifactId>SparkStreamingDemo</artifactId><version>1.0-SNAPSHOT</version><properties><maven.compiler.source>8</maven.compiler.source><maven.compiler.target>8</maven.compiler.target><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding></properties><dependencies><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>3.3.0</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>3.3.0</version></dependency><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.12</artifactId><version>3.3.0</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.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%ex

1.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输出操作

Spark Streaming编程基础

文章目录 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…...

编程日记 2025/5/26 13:00:50

android wifi AsyncChannel（WifiManager和WifiP2pManager）

AynscChannel的讲解 [Android]AsyncChannel介绍-CSDN博客 WifiP2pManager里的channel的使用理解 WifiP2pManager.java public void createGroup(Channel c, ActionListener listener) {checkChannel(c);c.mAsyncChannel.sendMessage(CREATE_GROUP, WifiP2pGroup.NETWORK_ID_PE…...

编程日记 2025/5/26 6:41:36

生存网络与mlr3proba

在R语言中，mlr3包是一个用于机器学习的强大工具包。它提供了一种简单且灵活的方式来执行超参数调整。生存网络是一种用于生存分析的模型，常用在医学和生物学领域。生存分析是一种统计方法，用于研究事件发生的时间和相关因素对事件发生的影响。生存网络可以用来预测个体在给…...

编程日记 2025/5/26 11:51:33

C#Object类型的索引，序列化和反序列化

前言最近在编写一篇关于标准Mes接口框架的文章。其中有一个非常需要考究的内容时如果实现数据灵活和可使用性强。因为考虑数据灵活性，所以我一开始选取了Object类型作为数据类型，Object作为数据Value字段，String作为数据Key字段&#xff0c…...

编程日记 2025/5/26 6:41:10

【动态规划】记忆化搜索

Ban or Problem - A - Codeforces 【CCPC】2022年绵阳站部分题解（ACGM）_ban or pick, whats the trick-CSDN博客 #include<iostream> using namespace std; #include<cstring> #include<algorithm> #define inf -0x3f3f3f3f #defi…...

编程日记 2025/5/26 12:13:22

深度学习 DAY1：RNN 神经网络及其变体网络（LSTM、GRU）

实验介绍 RNN 网络是一种基础的多层反馈神经网络，该神经网络的节点定向连接成环，其内部状态可以展示动态时序行为。相比于前馈神经网络，该网络内部具有很强的记忆性，它可以利用它内部的记忆来处理任意时序的输入序列，…...

编程日记 2025/5/26 12:41:10

BW复制ERP数据源跑程序激活后才可见

场景： BW提取ERP数据走ODP通道之后，数据源需要用下列程序激活加入白名单后才能被BW系统访问到； 检查： 1、RSA6检查数据源是否可正常使用，若为绿√表示可正常访问，反之，则不行。 2、白名单表ROO…...

编程日记 2025/5/26 12:03:36

MarsCode青训营打卡Day10（2025年1月23日）|稀土掘金-147.寻找独一无二的糖葫芦串、119.游戏队友搜索

资源引用： 147.寻找独一无二的糖葫芦串 119.游戏队友搜索今日小记： 回乡聚会陪家人，休息一天~ 稀土掘金-147.寻找独一无二的糖葫芦串（147.寻找独一无二的糖葫芦串） 题目分析： 给定n个长度为m的字符串表…...

编程日记 2025/5/26 12:14:37

无人机 PX4 飞控 | PX4源码添加自定义参数方法并用QGC显示与调整

无人机 PX4 飞控 | PX4源码添加自定义参数方法并用QGC显示与调整 0 前言之前文章添加了一个自定义的模块，本篇文章在之前的自定义模块中，添加两个自定义参数使用QGC显示出来，并通过QGC调整参数值，代码实现参数更新新增的参…...

编程日记 2025/5/26 12:09:25

【全栈】SprintBoot+vue3迷你商城-扩展：vue3项目创建及目录介绍

【全栈】SprintBootvue3迷你商城-扩展：vue3项目创建及目录介绍往期的文章都在这里啦，大家有兴趣可以看一下【全栈】SprintBootvue3迷你商城（1） 【全栈】SprintBootvue3迷你商城（2） 【全栈】SprintBootvu…...

编程日记 2025/5/26 12:21:29

Linux系统 C/C++编程基础——基于Qt的图形用户界面编程

ℹ️大家好，我是练小杰，今天周四了，距离除夕只有4天了，各位今年卫生都搞完了吗！😆 本文是接着昨天Linux 系统C/C编程的知识继续讲，基于Qt的图形用户界面编程概念及其命令，后续会不断…...

编程日记 2025/5/26 12:06:36

23.日常算法

1. 最小绝对差题目来源给你个整数数组 arr，其中每个元素都不相同。请你找到所有具有最小绝对差的元素对，并且按升序的顺序返回。每对元素对 [a,b] 如下： a , b 均为数组 arr 中的元素 a < b b - a 等于 arr 中任意两个元素的最小绝对…...

编程日记 2025/5/26 11:05:23

迅为RK3568开发板篇OpenHarmony实操HDF驱动控制LED-添加内核编译

编译内核时将该 HDF 驱动编译到镜像中，接下来编写驱动编译脚本 Makefile，代码如下所示： 加入编译体系，填加模块目录到 drivers/hdf_core/adapter/khdf/linux/Makefile 文件更多内容可以关注：迅为RK3568开发板篇OpenHa…...

编程日记 2025/5/26 10:49:03

为AI聊天工具添加一个知识系统之54 为事务处理设计基于DDD的一个 AI操作系统来处理维度

本文要点要点 Architecture程序它被设计为一个双面神结构的控制器，它的两侧一侧编译执行另一侧解释执行，自已则是一个翻译器--通过提供两个不同取向之间的结构映射的显示器（带图形用户接口GUI和命令行接口CLI 两种接口&#xff09…...

编程日记 2025/5/26 11:10:36

Golang 中除了加锁还有哪些安全读写共享变量的方式？

Golang 中除了加锁还有哪些安全读写共享变量的方式？ 在 Golang 中，除了使用 Mutex 锁来保护共享变量外，还可以通过 Channel 和原子性操作来实现安全读写共享变量。 1. 使用 Channel 原理 Channel 是 Golang 中用于 Goroutine 之间通信的…...

编程日记 2025/5/26 11:19:37

【优选算法】8----四数之和

有看过我上篇算法博客并且去做过的铁子们，对这道题的话应该就不会那么陌生了，因为这两道题的解题思路有着异曲同工之妙~ -----------------------------------------begin------------------------------------- 题目解析： 跟三数之和就多了…...

编程日记 2025/5/26 12:05:03

订单状态定时处理、来单提醒和客户催单（day10）

Spring Task 介绍 Spring Task 是 Spring 框架提供的任务调度工具，可以按照约定的时间自动执行某个代码逻辑。定位： 定时任务框架作用： 定时自动执行某段Java代码为什么要在 Java 程序中使用 Spring Task？ 应用场景&#xff1…...

编程日记 2025/5/21 17:10:26

备赛蓝桥杯之第十五届职业院校组省赛第一题：智能停车系统

提示：本篇文章仅仅是作者自己目前在备赛蓝桥杯中，自己学习与刷题的学习笔记，写的不好，欢迎大家批评与建议由于个别题目代码量与题目量偏大，请大家自己去蓝桥杯官网【连接高校和企业 - 蓝桥云课】去寻找原题&#xff0…...

编程日记 2025/5/25 23:27:56

【2024年华为OD机试】(C卷,100分)- 查找接口成功率最优时间段（JavaScriptJava PythonC/C++）

一、问题描述题目解析题目描述服务之间交换的接口成功率作为服务调用关键质量特性，某个时间段内的接口失败率使用一个数组表示。数组中每个元素都是单位时间内失败率数值，数组中的数值为 0~100 的整数。给定一个数值 minAverageLost，表…...

编程日记 2025/5/26 11:34:00

Linux进度条实现

Linux进度条实现 1.\r\n2.缓冲区3.缓冲区分类4.进度条实现 🌟🌟hello，各位读者大大们你们好呀🌟🌟 🚀🚀系列专栏：【Linux的学习】 📝📝本篇内容：\…...

编程日记 2025/5/26 3:55:16

Java如何实现反转义

Java如何实现反转义前提最近做的一个需求，是热搜词增加换一批的功能。功能做完自测后，交给了测试伙伴，但是测试第二天后就提了一个bug，出现了未知词 levis。第一眼看着像公司售卖的一个品牌-李维斯。然后再扒前人写的代码&…...

编程日记 2025/5/26 11:19:36

计算机网络（57）改进“尽最大努力交付”的服务

前言计算机网络中的“尽最大努力交付”服务是网络层的一种数据传输方式。这种服务的特点是网络层只负责尽力将数据报从源端传输到目的端，而不保证数据传输的可靠性。一、标记与分类为数据分组打上标记： 给不同性质的分组打上不同的标记&#x…...

编程日记 2025/5/26 12:02:11

Java 基于 SpringBoot 的校园外卖点餐平台微信小程序（附源码，部署，文档）

博主介绍：✌程序员徐师兄、7年大厂程序员经历。全网粉丝12w、csdn博客专家、掘金/华为云/阿里云/InfoQ等平台优质作者、专注于Java技术领域和毕业项目实战✌ 🍅文末获取源码联系🍅 👇🏻 精彩专栏推荐订阅👇…...

编程日记 2025/5/26 11:38:27

重构（4）

（一）添加解释性变量，使得代码更容易理解，更容易调试，也可以方便功能复用解释性的变量总价格为商品总价（单价*数量）-折扣（超过100个以上的打9折）邮费（原价的…...

编程日记 2025/5/26 12:16:02

【Arduino】语言参考功能

前言翻译Arduino 参考处列出的常用函数。文中为了减少篇幅，达到能快速翻到查询的目标，在介绍函数中，对部分内容进行了省略，不会列出函数输入参数类型，以及使用注意事项等等，所以若是首次使用或者是调试时出…...

编程日记 2025/5/25 18:46:57

CMake使用CPack制作安装程序

CPack的功能很强大，笔者前面有一博文使用CMake的CPack工具打包项目介绍了一下使用CPack来打包成7z压缩文件，不仅如此，它还可以生成各平台的安装包。 CPack支持以下类型的生成器： 名称文件类型平台及说明STGZSTGZ(.sh)自解压文件…...

编程日记 2025/5/26 11:57:34

Flink运行时架构

一、系统架构 1）作业管理器（JobManager） JobManager是一个Flink集群中任务管理和调度的核心，是控制应用执行的主进程。也就是说，每个应用都应该被唯一的JobManager所控制执行。 JobManger又包含3个不同的组件。 &am…...

编程日记 2025/5/25 1:05:22

平衡二叉树(力扣110)

所谓平衡二叉树，就是每一个节点的左右子树的高度差不大于1。而一个子树的高度，就是父节点的最大高度。这道题的思路其实和二叉树的最大深度(力扣104)-CSDN博客有很大的相似之处，都需要将左右子树的高度返回给父节点，因此也是采用后…...

编程日记 2025/5/26 11:30:12

【玩转全栈】---基于YOLO8的图片、视频目标检测

本篇主要讲YOLO8的具体操作，想要了解YOLO的具体原理，可以去官网查询目录下载ultralytics库开始检测介绍 YOLOv8（You Only Look Once Version 8）是 YOLO 系列的最新版本，由 Ultralytics 开发并发布，是一…...

编程日记 2025/5/26 11:56:01

ES6+新特性，var、let 和 const 的区别

在 JavaScript 中，var、let 和 const 都用于声明变量，但它们有一些重要的区别，主要体现在作用域、可变性和提升机制等方面。 1. 作用域（Scope） var: var 声明的变量是函数作用域，也就是说，它…...

编程日记 2025/5/26 11:09:21

汇编实验·分支程序设计

一、实验目的： 1.能够熟练的进行分支程序的编写，掌握条件语句对应的汇编语言指令的实现 2.掌握多个条件的分支语句的实现原理，理解C语言中的逻辑运算“短路”特征二、实验内容 1.对2和3任务中的C代码在VS2022中运行，设置生成对应的汇编代码，观察生成代码的不同，着重…...

编程日记 2025/5/25 21:04:54

激光线扫相机无2D图像的标定方案

方案一：基于运动控制平台的标定适用场景：若激光线扫相机安装在可控运动平台（如机械臂、平移台、旋转台）上，且平台的运动精度已知（例如通过编码器或高精度步进电机控制）。步骤： 标…...

编程日记 2025/5/26 11:02:44

【Python・机器学习】多元回归模型（原理及代码）

前言自学笔记，分享给语言学/语言教育学方向的，但对语言数据处理感兴趣但是尚未入门，却需要在论文中用到的小伙伴，欢迎大佬们补充或绕道。ps：本文最少限度涉及公式讲解（文科生小白友好体质）&am…...

编程日记 2025/5/26 12:20:13

ubuntu20.04安装使用direct_visual_lidar_calibration标定雷达和相机

官方链接GitHub - koide3/direct_visual_lidar_calibration: A toolbox for target-less LiDAR-camera calibration [ROS1/ROS2] 官方安装方式 Installation - direct_visual_lidar_calibration 安装依赖 sudo apt install libomp-dev libboost-all-dev libglm-dev libglfw…...

编程日记 2025/5/21 3:27:16

Android 自定义View时四个构造函数使用详解

该文章我们以自定义View继承TextView为例来讲解创建自定义View命名MyTextView，并使其继承TextView 1、自定义View时第一个构造函数 // 第一个构造函数主要是在Java代码中声明一个MyTextView时所用 // 类似这种(MyTextView myTextViewnew MyTextView(this);) // 不…...

编程日记 2025/5/24 21:59:18

linux中关闭服务的开机自启动

引言 systemctl 是 Linux 系统中用于管理 systemd 服务的命令行工具。它可以用来启动、停止、重启服务，管理服务的开机自启动，以及查看服务的状态等。什么是 systemd？ systemd 是现代 Linux 发行版中默认的初始化系统（init sys…...

编程日记 2025/5/26 5:50:32

【go语言】go的卸载与安装

一、卸载go sudo rm -rf /usr/local/go sudo apt-get remove golang sudo apt-get remove golang-go sudo apt-get autoremove wget https://dl.google.com/go/go1.19.linux-amd64.tar.gz sudo tar -xzf go1.19.linux-amd64.tar.gz -C /usr/local go env -w GOPROXY"http…...

编程日记 2025/5/23 3:49:20

微软Win10 RP 19045.5435(KB5050081)预览版发布!

系统之家1月20日最新报道，微软面向Release Preview频道的Windows Insider项目成员，发布了适用于Windows10 22H2版本的KB5050081更新，更新后系统版本号将升至19045.5435。本次更新增加了对GB18030-2022标准的支持，同时新版日历将为…...

编程日记 2025/5/20 5:37:05

VUE elTree 无子级隐藏展开图标

这4个并没有下级节点，即它并不是叶子节点，就不需求展示前面的三角展开图标! 查阅官方文档如下描述，支持bool和函数回调处理，这里咱们选择更灵活的函数回调实现。给el-tree结构配置一下props，注意！ :pr…...

编程日记 2025/5/23 4:02:05

python实现代理服务器转换

使用 http.server 模块和 requests 库来实现，代理服务器可拦截匹配指定路径的 HTTP 请求，并打印或存储请求头和响应体。实现目标： 1、创建一个HTTP代理服务器。 2、代理所有流量，而对指定的接口（如 /api/target&#…...

编程日记 2025/5/26 7:19:46

计算机网络三张表（ARP表、MAC表、路由表）总结

参考： 网络三张表：ARP表, MAC表, 路由表，实现你的网络自由！！_mac表、arp表、路由表-CSDN博客网络中的三张表：ARP表、MAC表、路由表首先要明确一件事，如果一个主机要发送数据，那么必…...

编程日记 2025/5/25 17:13:28

Dangerzone：免费的危险的文件转换安全程序

我们知道很多杀毒软件会将很多文件视作恶意程序，尤其是带宏病毒的DOC文件，直接删除难免会带来不必要的麻烦，那么推荐一款免费的危险的文件转换安全程序：Dangerzone 。它可以检查各种可能有危险的文件并将其转换为安全版本&#xf…...

编程日记 2025/5/24 15:33:37

2025/1/21 学习Vue的第四天

睡觉。 --------------------------------------------------------------------------------------------------------------------------------- 11.Object.defineProperty 1.在我们之前学习JS的时候，普通得定义一个对象与属性。 <!DOCTYPE html> <h…...

编程日记 2025/5/23 16:46:22