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版本,首先在Maven pom文件中导…...
《Java初阶数据结构》----4.<线性表---Stack栈和Queue队列>
前言 大家好,我目前在学习java。之前也学了一段时间,但是没有发布博客。时间过的真的很快。我会利用好这个暑假,来复习之前学过的内容,并整理好之前写过的博客进行发布。如果博客中有错误或者没有读懂的地方。热烈欢迎大家在评论区…...
Android SurfaceFlinger——关联EGL三要素(二十七)
通过前面的文章我们得到了 EGL 的三要素——Display、Surface 和 Context。其中,Display 是一个图形显示系统或者硬件屏幕,Surface 代表一个可以被渲染的图像缓冲区,Context 包含了 OpenGL ES 的状态信息和资源,它是执行 OpenGL 命令的环境。下一步就是调用 eglMakeCurrent…...
Unity3D之TCP网络通信(客户端)
文章目录 概述TCP核心类异步机制 Unity中创建TCP客户端Unity中其它脚本获取TCP客户端接受到的数据后续改进 本文将以Unity3D应用项目作为客户端去连接制定的服务器为例进行相关说明。 Unity官网参考资料: https://developer.unity.cn/projects/6572ea1bedbc2a001ef…...
Kotlin 中 标准库函数
在 Kotlin 中,标准库提供了许多实用的函数,这些函数可以帮助简化代码、提高效率,以下是一些常用的标准库函数及其功能: let: let 函数允许你在对象上执行一个操作,并返回结果。它通常与安全调用操作符 ?. 一起使用&a…...
【教学类-69-01】20240721铠甲勇士扑克牌(随机14个数字+字母)涂色(男孩篇)
背景需求: 【教学类-68-01】20240720裙子涂色(女孩篇)-CSDN博客文章浏览阅读250次。【教学类-68-01】20240720裙子涂色(女孩篇)https://blog.csdn.net/reasonsummer/article/details/140578153 前期制作了女孩涂色延…...
Adobe“加速”创意人士开启设计新篇章
近日,Adobe公司宣布了其行业领先的专业设计应用程序——Adobe Illustrator和Adobe Photoshop的突破性创新。这一重大更新不仅为创意专业人士带来了前所未有的设计可能性和工作效率提升,还让不论是插画师、设计师还是摄影师,都能从中受益并创作…...
释疑 803-(1)概述 精炼提纯版
目录 习题 1-01计算机网络可以向用户提供哪些服务? 1-02 试简述分组交换的要点。 1-03 试从多个方面比较电路交换、报文交换和分组交换的主要优缺点。 1-05 互联网基础结构的发展大致分为哪几个阶段?请指出这几个阶段最主要的特点。 1-06 简述互联网标准制定的几个阶段…...
人工智能与机器学习原理精解【6】
文章目录 数值优化基础理论凹凸性定义在国外与国内存在不同国内定义国外定义总结示例与说明注意事项 国内凹凸性二阶定义的例子凹函数例子凸函数例子 凸函数(convex function)的开口方向凸函数的二阶导数凸函数的二阶定义单变量函数的二阶定义多变量函数…...
JDK、JRE、JVM之间的关系
JDK是Java的开发环境,用JDK开发了JAVA程序后,通过JDK中的编译程序(javac)将java文件编译成字节码文件,作为运行环境的JRE,字节码文件在JRE上运行,作为虚拟机的JVM解析这些字节码,映射…...
redis构建集群时,一直Waiting for the cluster to join
redis构建集群时,一直Waiting for the cluster to join 前置条件参考 前置条件 这是我搭建的集群相关信息,三台虚拟机,分别是一主一从。在将所有虚拟机中redis服务器用到的tcp端口都打开之后,进行构建集群。但是出现上面的情况。 …...
C++之类与对象(2)
前言 今天将步入学习类的默认成员函数,本节讲解其中的构造函数和析构函数。 1.类的默认成员函数 在 C 中,如果一个类没有显式定义某些成员函数,编译器会自动为该类生成默认的成员函数。以下是编译器可能会生成的默认成员函数: 默…...
「树形结构」基于 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,您可以轻松获取较新的软件版本或官方 Ubuntu 存储库无法提供的软件。 为什么使用PPA? 正如…...
20240724-然后用idea创建一个Java项目/配置maven环境/本地仓储配置
1.创建一个java项目 (1)点击页面的create project,然后next (2)不勾选,继续next (3)选择新项目名称,新项目路径,然后Finsh,在新打开的页面选择…...
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 题(二)
✍个人博客:Pandaconda-CSDN博客 📣专栏地址:http://t.csdnimg.cn/UWz06 📚专栏简介:在这个专栏中,我将会分享 Golang 面试中常见的面试题给大家~ ❤️如果有收获的话,欢迎点赞👍收藏…...
状态模式与订单状态机的实现
状态模式 状态模式(State Design Pattern)是一种行为设计模式,用于在对象的内部状态改变时改变其行为。这种模式可以将状态的变化封装在状态对象中,使得对象在状态变化时不会影响到其他代码,提升了代码的灵活性和可维…...
【MSP430】MSP430是什么?与STM32对比哪个性能更佳?
一、MSP430是什么? MSP430F5529LP是一款由德州仪器(TI)推出的16位微控制器单元(MCU)开发板,具有USB功能,内存配置为128KB闪存和8KB RAM,工作频率高达25MHz。 这款MCU以其高性能和多…...
Win11 操作(四)g502鼠标连接电脑不亮灯无反应
罗技鼠标连接电脑不亮灯无反应 前言 罗技技术💩中💩,贴吧技术神中神! 最近买了一个g502,结果买回来直接插上电脑连灯都不亮,问了一下客服。客服简单的让我换接口,又是下载ghub之类的…...
自定义QDialog使用详解
自定义QDialog使用详解 一、创建 QDialog 对象二、QDialog设置布局三、QDialog控制模态行为3.1 模态和非模态区别3.2 QDialog的模态使用四、使用 QDialogButtonBox五、处理对话框的结果六、使用 QDialog 的信号和槽QDialog是Qt框架中用于创建对话框窗口的基本类。对话框窗口通常…...
Pytorch使用教学2-Tensor的维度
在PyTorch使用的过程中,维度转换一定少不了。而PyTorch中有多种维度形变的方法,我们该在什么场景下使用什么方法呢? 本小节我们使用的张量如下: # 一维向量 t1 torch.tensor((1, 2)) # 二维向量 t2 torch.tensor([[1, 2, 3], …...
Interesting bug caused by getattr
题意:由 getattr 引起的有趣的 bug 问题背景: I try to train 8 CNN models with the same structures simultaneously. After training a model on a batch, I need to synchronize the weights of the feature extraction layers in other 7 models. …...
获取后端返回的图形验证码
如果后端返回的直接就是一个图形,有以下几种方式展示 一、直接在img标签里面的src里面调用接口 <img :src"dialogSrc" class"photo" alt"验证码图片" click"changeDialog">let orgUrl "/api/captcha" …...
奇怪的Excel单元格字体颜色格式
使用VBA代码修改单元格全部字符字体颜色是个很简单的任务,例如设置A1单元格字体颜色为红色。 Range("A1").Font.Color RGB(255, 0, 0)有时需要修改部分字符的颜色,如下图所示,将红色字符字体颜色修改为蓝色。代码将会稍许复杂&am…...
浅谈芯片验证中的仿真运行之 timescale (五)提防陷阱
一 仿真单位 timeunit 我们知道,当我们的代码中写清楚延时语句时,若不指定时间单位,则使用此单位; 例如: `timescale 1ns/1ps 则 #15 语句表示delay15ns; 例:如下代码,module a 的timescale是1ns/1ps, module b 是1ps/1ps; module b中的clk,频率是由输入参…...
uniapp 重置表单数据
场景 例如有数据如下 data(){return {queryForm:{value1:undefined,}} } 点击重置时候想重置form的数据, 操作 Object.assign(this.$data.queryForm, this.$options.data().queryForm); 就可以重置数据...
自学YOLO前置知识
YOLO前置知识 学习YOLO(You Only Look Once)之前,掌握一些前置知识会帮助你更好地理解和应用该技术。以下是一些推荐的前置知识领域: 计算机视觉基础: 图像处理:了解图像的基本处理技术,如滤波…...
Ubuntu18.04 编译报错: Could NOT find JNI
一、问题描述 Ubuntu18.04 编译报错 OpenCV 时,出现以下错误: Could NOT find JNI (missing: JAVA_INCLUDE_PATH JAVA_INCLUDE_PATH2 JAVA_AWT_INCLUDE_PATH)二、解决方法 先执行以下指令, export JAVA_HOME/usr/lib/jvm/java-8-openjdk-am…...
SQL labs-SQL注入(五,使用sqlmap进行cookie注入)
本文仅作为学习参考使用,本文作者对任何使用本文进行渗透攻击破坏不负任何责任。 引言: Cookie 是一些数据, 存储于你电脑上的文本文件中。当 web 服务器向浏览器发送 web 页面时,在连接关闭后,服务端不会记录用户的信息。Cookie…...
C语言——内存管理
目录 前言 一、内存分类 1. 栈区(Stack) 2. 堆区(Heap) 3. 数据段(Data Segment) 4. 代码段(Code Segment) 二、内存分配方式 1、静态内存分配 2、栈内分配 3、动态内存分配 &#x…...
Unity UGUI 之 Image和Rawimage
本文仅作学习笔记与交流,不作任何商业用途 本文包括但不限于unity官方手册,唐老狮,麦扣教程知识,引用会标记,如有不足还请斧正 1.Image是什么 Unity - 手册:图像 精灵格式是什么? 1.2重要参数 …...
Lua 语法学习笔记
Lua 语法学习笔记 安装(windows) 官网:https://www.lua.org/ 下载SDK 解压&修改名称(去除版本号) 将lua后面的版本号去掉,如lua54.exe->lua.ext 配置环境变量 数据类型 数据类型描述nil这个最简单,只有值n…...
Prometheus配置alertmanager告警
1、拉取镜像并运行 1、配置docker镜像源 [rootlocalhost ~]# vim /etc/docker/daemon.json {"registry-mirrors": ["https://dfaad.mirror.aliyuncs.com"] } [rootlocalhost ~]# systemctl daemon-reload [rootlocalhost ~]# systemctl restart docker2、…...
.net core 外观者设计模式 实现,多种支付选择
1,接口 /// <summary>/// Web页面支付/// </summary>public interface IWebPagePay{public WebPagePayResult CreatePay(string productName, string orderSn, string totalPrice);}2,实现接口 实现阿里支付 public class AliPagePay : IWe…...
Matlab 命令行窗口默认输出(异常)
目录 前言Matlab 先验知识1 异常输出的代码2 正常输出的代码 前言 在单独调试 Matlab 写的函数时出现不想出现的异常打印值,逐个注释排查才找到是 if elseif else 代码块的问题,会默认打印输出 else 部分第一个返回值的值(下方代码中的 P值&…...
LeetCode/NowCoder-二叉树OJ练习
励志冰檗:形容在清苦的生活环境中激励自己的意志。💓💓💓 目录 说在前面 题目一:单值二叉树 题目二:相同的树 题目三:对称二叉树 题目四:二叉树的前序遍历 题目五:另…...
PSINS工具箱函数介绍——insplot
insplot是一个绘图命令,用于将avp数据绘制出来 本文所述的代码需要基于PSINS工具箱,工具箱的讲解: PSINS初学指导基于PSINS的相关程序设计(付费专题)使用方法 此函数使用起来也很简单,直接后面加avp即可,如: insplot(avp);其中,avp为: 每行表示一个时间1~3列为姿态…...
Docker简单快速入门
1. 安装Docker 基于 Ubuntu 24.04 LTS 安装Docker 。 # 更新包索引并安装依赖包 sudo apt-get update sudo apt-get install -y apt-transport-https ca-certificates curl software-properties-common# 添加Docker的官方GPG密钥并存储在正确的位置 curl -fsSL https://mirror…...
【2024最新华为OD-C/D卷试题汇总】[支持在线评测] 图像物体的边界(200分) - 三语言AC题解(Python/Java/Cpp)
🍭 大家好这里是清隆学长 ,一枚热爱算法的程序员 ✨ 本系列打算持续跟新华为OD-C/D卷的三语言AC题解 💻 ACM银牌🥈| 多次AK大厂笔试 | 编程一对一辅导 👏 感谢大家的订阅➕ 和 喜欢💗 🍿 最新华为OD机试D卷目录,全、新、准,题目覆盖率达 95% 以上,支持题目在线…...
【无人机】低空经济中5G RedCap芯片的技术分析报告
1. 引言 图一. 新基建:低空经济 低空经济作为一种新兴的经济形态,涵盖了无人机、电动垂直起降飞行器(eVTOL)、低空物流、空中交通管理等多个领域。随着5G网络的普及和演进,5G RedCap(Reduced Capability&a…...
MongoDB教程(二十一):MongoDB大文件存储GridFS
💝💝💝首先,欢迎各位来到我的博客,很高兴能够在这里和您见面!希望您在这里不仅可以有所收获,同时也能感受到一份轻松欢乐的氛围,祝你生活愉快! 文章目录 引言一、GridFS…...
vue 搜索框
效果 创建搜索组件: 在Vue项目中,首先需要创建一个搜索组件。这个组件通常包含一个输入框和一个搜索按钮。使用v-model指令将输入框与组件的数据属性(如searchKeyword)进行双向绑定,以便获取用户输入的关键词。处理搜索…...
国科大作业考试资料-人工智能原理与算法-2024新编-第五次作业整理
1、本题以井字棋(圈与十字游戏)为例练习博弈中的基本概念。定义X_n为恰好有n个X而没有O 的行、列或者对角线的数目。同样O_n为正好有n 个O的行、列或者对角线的数目。效用函数给 X_3=1的棋局+1, 给O_3=1的棋局-1。所有其他终止状态效用值为0。对于非终止状态,使用线性的 …...
C++五子棋(未做完,但能玩,而且还不错)
代码放下面了,关于步骤介绍的我以后再完善一下。 #include<bits/stdc.h> #include<cstdio> #include<cstdlib> #include<ctime> #include<windows.h> #include<stdlib.h> #include<time.h> #define random(x) (rand()%x…...
二分查找代码详解
二分查找代码实现 以下是完整的代码和解释: #include <stdio.h>int binarySearch(int arr[], int length, int target) {int left 0;int right length - 1;while (left < right) {int mid left (right - left) / 2; // 防止溢出if (arr[mid] target…...
uniapp的h5,读取本地txt带标签的文件
效果图 使用的回显的标签是u-parse,下面的网址讲了这个标签的相关 https://www.cnblogs.com/huihuihero/p/12978903.html 导入此插件 https://ext.dcloud.net.cn/plugin?id364 使用 uni.request({// 本地文件url: "/static/互联网医院医师端用户协议.txt…...
韦东山嵌入式linux系列-具体单板的按键驱动程序(查询方式)
1 GPIO 操作回顾 (1)使能模块; (2)设置引脚的模式(工作于GPIO模式); (3)设置GPIO本身(输入/输出); (4&…...
如何使用 API list 极狐GitLab 群组中的镜像仓库?
GitLab 是一个全球知名的一体化 DevOps 平台,很多人都通过私有化部署 GitLab 来进行源代码托管。极狐GitLab :https://gitlab.cn/install?channelcontent&utm_sourcecsdn 是 GitLab 在中国的发行版,专门为中国程序员服务。可以一键式部署…...
PHP设计模式-简单工厂模式
核心: 一、定义一个接口类里面写规定好的方法。 interface Message{public function send(array $params);public function getMessage(array $params);public function getCode(array $params);} 二、定义产品类 、产品类继承接口类 class AlliYunSms implements …...