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

Spark实时(三):Structured Streaming入门案例

文章目录

Structured Streaming入门案例

一、Scala代码如下

二、Java 代码如下

三、以上代码注意点如下


Structured Streaming入门案例

我们使用Structured Streaming来监控socket数据统计WordCount。这里我们使用Spark版本为3.4.3版本,首先在Maven pom文件中导入以下依赖:

 <!-- 配置以下可以解决 在jdk1.8环境下打包时报错 “-source 1.5 中不支持 lambda 表达式” --><properties><project.build.sourceEncoding>UTF-8</project.build.sourceEncoding><maven.compiler.source>1.8</maven.compiler.source><maven.compiler.target>1.8</maven.compiler.target><spark.version>3.4.3</spark.version></properties><dependencies><!-- Spark-core --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.12</artifactId><version>${spark.version}</version></dependency><!-- SparkSQL --><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.12</artifactId><version>${spark.version}</version></dependency><!-- SparkSQL  ON  Hive--><dependency><groupId>org.apache.spark</groupId><artifactId>spark-hive_2.12</artifactId><version>${spark.version}</version></dependency><!--mysql依赖的jar包--><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>5.1.47</version></dependency><!--SparkStreaming--><dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.12</artifactId><version>${spark.version}</version></dependency><!-- Kafka 0.10+ Source For Structured Streaming--><dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql-kafka-0-10_2.12</artifactId><version>${spark.version}</version></dependency><!-- 向kafka 生产数据需要包 --><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.8.0</version></dependency><!-- Scala 包--><dependency><groupId>org.scala-lang</groupId><artifactId>scala-library</artifactId><version>2.12.15</version></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-compiler</artifactId><version>2.12.15</version></dependency><dependency><groupId>org.scala-lang</groupId><artifactId>scala-reflect</artifactId><version>2.12.15</version></dependency><dependency><groupId>log4j</groupId><artifactId>log4j</artifactId><version>1.2.12</version></dependency><dependency><groupId>com.google.collections</groupId><artifactId>google-collections</artifactId><version>1.0</version></dependency></dependencies>

一、Scala代码如下

package com.lanson.structuredStreaming/***  Structured Streaming 实时读取Socket数据*/import org.apache.spark.sql.streaming.StreamingQuery
import org.apache.spark.sql.{DataFrame, Dataset, SparkSession}/*** Structured Streaming 读取Socket数据*/
object SSReadSocketData {def main(args: Array[String]): Unit = {//1.创建SparkSession对象val spark: SparkSession = SparkSession.builder().master("local").appName("StructuredSocketWordCount")//默认200个并行度,由于源头数据量少,可以设置少一些并行度.config("spark.sql.shuffle.partitions",1).getOrCreate()import spark.implicits._spark.sparkContext.setLogLevel("Error")//2.读取Socket中的每行数据,生成DataFrame默认列名为"value"val lines: DataFrame = spark.readStream.format("socket").option("host", "node3").option("port", 9999).load()//3.将每行数据切分成单词,首先通过as[String]转换成Dataset操作val words: Dataset[String] = lines.as[String].flatMap(line=>{line.split(" ")})//4.按照单词分组,统计个数,自动多一个列countval wordCounts: DataFrame = words.groupBy("value").count()//5.启动流并向控制台打印结果val query: StreamingQuery = wordCounts.writeStream//更新模式设置为complete.outputMode("complete").format("console").start()query.awaitTermination()}}

 

二、Java 代码如下

package com.lanson.structuredStreaming;import java.util.Arrays;
import java.util.Iterator;
import java.util.concurrent.TimeoutException;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Encoders;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.streaming.StreamingQuery;
import org.apache.spark.sql.streaming.StreamingQueryException;public class SSReadSocketData01 {public static void main(String[] args) throws StreamingQueryException, TimeoutException {SparkSession spark = SparkSession.builder().master("local").appName("SSReadSocketData01").config("spark.sql.shuffle.partitions", 1).getOrCreate();spark.sparkContext().setLogLevel("Error");Dataset<Row> lines = spark.readStream().format("socket").option("host", "node3").option("port", 9999).load();Dataset<String> words = lines.as(Encoders.STRING()).flatMap(new FlatMapFunction<String, String>() {@Overridepublic Iterator<String> call(String line) throws Exception {return Arrays.asList(line.split(" ")).iterator();}}, Encoders.STRING());Dataset<Row> wordCounts = words.groupBy("value").count();StreamingQuery query = wordCounts.writeStream().outputMode("complete").format("console").start();query.awaitTermination();}
}

 

以上代码编写完成之后,在node3节点执行“nc -lk 9999”启动socket服务器,然后启动代码,向socket中输入以下数据:

第一次输入:a b c
第二次输入:d a c
第三次输入:a b c

可以看到控制台打印如下结果:

-------------------------------------------
Batch: 1
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    c|    1|
|    b|    1|
|    a|    1|
+-----+-----+-------------------------------------------
Batch: 2
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    d|    1|
|    c|    2|
|    b|    1|
|    a|    2|
+-----+-----+-------------------------------------------
Batch: 3
-------------------------------------------
+-----+-----+
|value|count|
+-----+-----+
|    d|    1|
|    c|    3|
|    b|    2|
|    a|    3|
+-----+-----+

三、以上代码注意点如下

  • SparkSQL 默认并行度为200,这里由于数据量少,可以将并行度通过参数“spark.sql.shuffle.partitions”设置少一些。
  • StructuredStreaming读取过来数据默认是DataFrame,默认有“value”名称的列
  • 对获取的DataFrame需要通过as[String]转换成Dataset进行操作
  • 结果输出时的OutputMode有三种输出模式:Complete Mode、Append Mode、Update Mode。

  • 📢博客主页:https://lansonli.blog.csdn.net
  • 📢欢迎点赞 👍 收藏 ⭐留言 📝 如有错误敬请指正!
  • 📢本文由 Lansonli 原创,首发于 CSDN博客🙉
  • 📢停下休息的时候不要忘了别人还在奔跑,希望大家抓紧时间学习,全力奔赴更美好的生活✨

相关文章:

Spark实时(三):Structured Streaming入门案例

文章目录 Structured Streaming入门案例 一、Scala代码如下 二、Java 代码如下 三、以上代码注意点如下 Structured Streaming入门案例 我们使用Structured Streaming来监控socket数据统计WordCount。这里我们使用Spark版本为3.4.3版本&#xff0c;首先在Maven pom文件中导…...

《Java初阶数据结构》----4.<线性表---Stack栈和Queue队列>

前言 大家好&#xff0c;我目前在学习java。之前也学了一段时间&#xff0c;但是没有发布博客。时间过的真的很快。我会利用好这个暑假&#xff0c;来复习之前学过的内容&#xff0c;并整理好之前写过的博客进行发布。如果博客中有错误或者没有读懂的地方。热烈欢迎大家在评论区…...

Android SurfaceFlinger——关联EGL三要素(二十七)

通过前面的文章我们得到了 EGL 的三要素——Display、Surface 和 Context。其中,Display 是一个图形显示系统或者硬件屏幕,Surface 代表一个可以被渲染的图像缓冲区,Context 包含了 OpenGL ES 的状态信息和资源,它是执行 OpenGL 命令的环境。下一步就是调用 eglMakeCurrent…...

Unity3D之TCP网络通信(客户端)

文章目录 概述TCP核心类异步机制 Unity中创建TCP客户端Unity中其它脚本获取TCP客户端接受到的数据后续改进 本文将以Unity3D应用项目作为客户端去连接制定的服务器为例进行相关说明。 Unity官网参考资料&#xff1a; https://developer.unity.cn/projects/6572ea1bedbc2a001ef…...

Kotlin 中 标准库函数

在 Kotlin 中&#xff0c;标准库提供了许多实用的函数&#xff0c;这些函数可以帮助简化代码、提高效率&#xff0c;以下是一些常用的标准库函数及其功能&#xff1a; let: let 函数允许你在对象上执行一个操作&#xff0c;并返回结果。它通常与安全调用操作符 ?. 一起使用&a…...

【教学类-69-01】20240721铠甲勇士扑克牌(随机14个数字+字母)涂色(男孩篇)

背景需求&#xff1a; 【教学类-68-01】20240720裙子涂色&#xff08;女孩篇&#xff09;-CSDN博客文章浏览阅读250次。【教学类-68-01】20240720裙子涂色&#xff08;女孩篇&#xff09;https://blog.csdn.net/reasonsummer/article/details/140578153 前期制作了女孩涂色延…...

Adobe“加速”创意人士开启设计新篇章

近日&#xff0c;Adobe公司宣布了其行业领先的专业设计应用程序——Adobe Illustrator和Adobe Photoshop的突破性创新。这一重大更新不仅为创意专业人士带来了前所未有的设计可能性和工作效率提升&#xff0c;还让不论是插画师、设计师还是摄影师&#xff0c;都能从中受益并创作…...

释疑 803-(1)概述 精炼提纯版

目录 习题 1-01计算机网络可以向用户提供哪些服务? 1-02 试简述分组交换的要点。 1-03 试从多个方面比较电路交换、报文交换和分组交换的主要优缺点。 1-05 互联网基础结构的发展大致分为哪几个阶段?请指出这几个阶段最主要的特点。 1-06 简述互联网标准制定的几个阶段…...

人工智能与机器学习原理精解【6】

文章目录 数值优化基础理论凹凸性定义在国外与国内存在不同国内定义国外定义总结示例与说明注意事项 国内凹凸性二阶定义的例子凹函数例子凸函数例子 凸函数&#xff08;convex function&#xff09;的开口方向凸函数的二阶导数凸函数的二阶定义单变量函数的二阶定义多变量函数…...

JDK、JRE、JVM之间的关系

JDK是Java的开发环境&#xff0c;用JDK开发了JAVA程序后&#xff0c;通过JDK中的编译程序&#xff08;javac&#xff09;将java文件编译成字节码文件&#xff0c;作为运行环境的JRE&#xff0c;字节码文件在JRE上运行&#xff0c;作为虚拟机的JVM解析这些字节码&#xff0c;映射…...

redis构建集群时,一直Waiting for the cluster to join

redis构建集群时&#xff0c;一直Waiting for the cluster to join 前置条件参考 前置条件 这是我搭建的集群相关信息&#xff0c;三台虚拟机&#xff0c;分别是一主一从。在将所有虚拟机中redis服务器用到的tcp端口都打开之后&#xff0c;进行构建集群。但是出现上面的情况。 …...

C++之类与对象(2)

前言 今天将步入学习类的默认成员函数&#xff0c;本节讲解其中的构造函数和析构函数。 1.类的默认成员函数 在 C 中&#xff0c;如果一个类没有显式定义某些成员函数&#xff0c;编译器会自动为该类生成默认的成员函数。以下是编译器可能会生成的默认成员函数&#xff1a; 默…...

「树形结构」基于 Antd 实现一个动态增加子节点+可拖拽的树

效果 如图所示 实现 import { createRoot } from react-dom/client; import React, { useState } from react; import { Tree, Input, Button } from antd; import { PlusOutlined } from ant-design/icons;const { TreeNode } Tree; const { Search } Input;const ini…...

ubuntu那些ppa源在哪

Ubuntu中的 PPA 终极指南 - UBUNTU粉丝之家 什么是PPA PPA 代表个人包存档。 PPA 允许应用程序开发人员和 Linux 用户创建自己的存储库来分发软件。 使用 PPA&#xff0c;您可以轻松获取较新的软件版本或官方 Ubuntu 存储库无法提供的软件。 为什么使用PPA&#xff1f; 正如…...

20240724-然后用idea创建一个Java项目/配置maven环境/本地仓储配置

1.创建一个java项目 &#xff08;1&#xff09;点击页面的create project&#xff0c;然后next &#xff08;2&#xff09;不勾选&#xff0c;继续next &#xff08;3&#xff09;选择新项目名称&#xff0c;新项目路径&#xff0c;然后Finsh&#xff0c;在新打开的页面选择…...

PaddleOCR-PP-OCRv4推理详解及部署实现(下)

目录 前言1. 检测模型1.1 预处理1.2 后处理1.3 推理 2. 方向分类器模型2.1 预处理2.2 后处理2.3 推理 3. 识别模型3.1 预处理3.2 后处理3.3 推理 4. PP-OCRv4部署4.1 源码下载4.2 环境配置4.2.1 配置CMakeLists.txt4.2.2 配置Makefile 4.3 ONNX导出4.4 engine生成4.4.1 检测模型…...

【Golang 面试基础题】每日 5 题(二)

✍个人博客&#xff1a;Pandaconda-CSDN博客 &#x1f4e3;专栏地址&#xff1a;http://t.csdnimg.cn/UWz06 &#x1f4da;专栏简介&#xff1a;在这个专栏中&#xff0c;我将会分享 Golang 面试中常见的面试题给大家~ ❤️如果有收获的话&#xff0c;欢迎点赞&#x1f44d;收藏…...

状态模式与订单状态机的实现

状态模式 状态模式&#xff08;State Design Pattern&#xff09;是一种行为设计模式&#xff0c;用于在对象的内部状态改变时改变其行为。这种模式可以将状态的变化封装在状态对象中&#xff0c;使得对象在状态变化时不会影响到其他代码&#xff0c;提升了代码的灵活性和可维…...

【MSP430】MSP430是什么?与STM32对比哪个性能更佳?

一、MSP430是什么&#xff1f; MSP430F5529LP是一款由德州仪器&#xff08;TI&#xff09;推出的16位微控制器单元&#xff08;MCU&#xff09;开发板&#xff0c;具有USB功能&#xff0c;内存配置为128KB闪存和8KB RAM&#xff0c;工作频率高达25MHz。 这款MCU以其高性能和多…...

Win11 操作(四)g502鼠标连接电脑不亮灯无反应

罗技鼠标连接电脑不亮灯无反应 前言 罗技技术&#x1f4a9;中&#x1f4a9;&#xff0c;贴吧技术神中神&#xff01; 最近买了一个g502&#xff0c;结果买回来直接插上电脑连灯都不亮&#xff0c;问了一下客服。客服简单的让我换接口&#xff0c;又是下载ghub之类的&#xf…...

利用快马ai快速生成流水线plc控制逻辑原型,无硬件也能验证思路

最近在做一个自动化流水线的小项目&#xff0c;需要设计PLC控制逻辑。传统方式需要先搭建硬件环境才能调试&#xff0c;但通过InsCode(快马)平台的AI辅助&#xff0c;我实现了无硬件环境下的快速原型验证&#xff0c;分享下这个实用经验。 项目背景与需求分析 这个流水线控制系…...

效率飙升:用快马平台一键生成项目模板,告别重复的vscode环境配置

作为一个经常需要在新设备上配置开发环境的前端开发者&#xff0c;我深刻体会到重复搭建项目的痛苦。每次换电脑或者开新项目&#xff0c;都要从头安装VSCode插件、配置构建工具、集成UI库……这些琐碎工作至少会浪费半小时。最近发现InsCode(快马)平台能完美解决这个问题&…...

DbGate数据库管理工具:Docker一键部署与跨平台远程访问实战

1. 为什么选择DbGateDocker组合 第一次接触DbGate是在一个需要同时管理MySQL和MongoDB的项目中。当时团队里有人用Navicat&#xff0c;有人用DBeaver&#xff0c;数据库类型切换时总要重新适应界面。直到发现这个支持多数据库的开源工具&#xff0c;才真正体会到什么叫"一…...

突破联想笔记本BIOS限制:LEGION BIOS高级设置工具全解析

突破联想笔记本BIOS限制&#xff1a;LEGION BIOS高级设置工具全解析 【免费下载链接】LEGION_Y7000Series_Insyde_Advanced_Settings_Tools 支持一键修改 Insyde BIOS 隐藏选项的小工具&#xff0c;例如关闭CFG LOCK、修改DVMT等等 项目地址: https://gitcode.com/gh_mirrors…...

跨平台网络资源嗅探下载工具:一站式解决多媒体内容获取难题

跨平台网络资源嗅探下载工具&#xff1a;一站式解决多媒体内容获取难题 【免费下载链接】res-downloader 资源下载器、网络资源嗅探&#xff0c;支持微信视频号下载、网页抖音无水印下载、网页快手无水印视频下载、酷狗音乐下载等网络资源拦截下载! 项目地址: https://gitcod…...

GitHub Desktop汉化终极指南:如何让英文界面瞬间变成中文

GitHub Desktop汉化终极指南&#xff1a;如何让英文界面瞬间变成中文 【免费下载链接】GitHubDesktop2Chinese GithubDesktop语言本地化(汉化)工具 项目地址: https://gitcode.com/gh_mirrors/gi/GitHubDesktop2Chinese 还在为GitHub Desktop的英文界面而头疼吗&#xf…...

技能组合玩法:OpenClaw串联百川2-13B-4bits与Stable Diffusion技能

技能组合玩法&#xff1a;OpenClaw串联百川2-13B-4bits与Stable Diffusion技能 1. 为什么需要技能组合&#xff1f; 去年我运营技术博客时&#xff0c;最头疼的就是内容生产的全流程管理。写一篇文章需要经历选题构思、文案撰写、配图制作、格式调整、发布上线等多个环节。每…...

从理论到实践:在快马平台构建基于openclaw的物流分拣仿真系统

最近在研究物流自动化分拣系统时&#xff0c;发现openclaw机械爪控制在实际应用中存在不少痛点。传统开发流程需要从零搭建仿真环境、编写控制逻辑、调试物理交互&#xff0c;整个过程耗时耗力。于是尝试用InsCode(快马)平台快速构建了一个物流分拣仿真系统&#xff0c;效果出乎…...

不止于读写:在HC32F460上为FATFS和SDIO驱动添加调试信息与性能测试

HC32F460深度优化&#xff1a;FATFS与SDIO驱动的调试技巧与性能压测实战 当你的HC32F460开发板已经能够读取SD卡文件时&#xff0c;真正的挑战才刚刚开始。那些隐藏在初始化失败、数据错位、速度瓶颈背后的秘密&#xff0c;往往需要更精密的调试手段才能揭开。本文将带你超越基…...

douyin-downloader:抖音视频批量下载解决方案

douyin-downloader&#xff1a;抖音视频批量下载解决方案 【免费下载链接】douyin-downloader 项目地址: https://gitcode.com/GitHub_Trending/do/douyin-downloader 在数字内容爆炸的时代&#xff0c;视频资源的高效管理已成为内容创作者、教育工作者和社交媒体运营者…...