SparkStreaming在实时处理的两个场景示例

简介

Spark Streaming是Apache Spark生态系统中的一个组件，用于实时流式数据处理。它提供了类似于Spark的API，使开发者可以使用相似的编程模型来处理实时数据流。

Spark Streaming的工作原理是将连续的数据流划分成小的批次，并将每个批次作为RDD（弹性分布式数据集）来处理。这样，开发者可以使用Spark的各种高级功能，如map、reduce、join等，来进行实时数据处理。Spark Streaming还提供了内置的窗口操作、状态管理、容错处理等功能，使得开发者能够轻松处理实时数据的复杂逻辑。

Spark Streaming支持多种数据源，包括Kafka、Flume、HDFS、S3等，因此可以轻松地集成到各种数据管道中。它还能够与Spark的批处理和SQL引擎进行无缝集成，从而实现流式处理与批处理的混合使用。
在这里插入图片描述

本文以 TCP、kafka场景讲解spark streaming的使用

消息队列下的信息铺抓

类似消息队列的有redis、kafka等核心组件。
本文以kafka为例，向kafka中实时抓取数据，

pom.xml中添加以下依赖

<dependencies>
    <!-- Spark Core -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-core_2.12</artifactId>
        <version>3.2.0</version>
    </dependency>

    <!-- Spark Streaming -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming_2.12</artifactId>
        <version>3.2.0</version>
    </dependency>

    <!-- Spark SQL -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-sql_2.12</artifactId>
        <version>3.2.0</version>
    </dependency>

    <!-- Kafka -->
    <dependency>
        <groupId>org.apache.kafka</groupId>
        <artifactId>kafka-clients</artifactId>
        <version>2.8.0</version>
    </dependency>

    <!-- Spark Streaming Kafka Connector -->
    <dependency>
        <groupId>org.apache.spark</groupId>
        <artifactId>spark-streaming-kafka-0-10_2.12</artifactId>
        <version>3.2.0</version>
    </dependency>

    <!-- PostgreSQL JDBC -->
    <dependency>
        <groupId>org.postgresql</groupId>
        <artifactId>postgresql</artifactId>
        <version>42.2.24</version>
    </dependency>
</dependencies>

创建项目编写以下代码实现功能

package org.example;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;

public class SparkStreamingKafka {
    public static void main(String[] args) throws InterruptedException {
        // 创建 Spark 配置
        SparkConf sparkConf = new SparkConf()
                .setAppName("spark_kafka")
                .setMaster("local[*]")
                .setExecutorEnv("setLogLevel", "ERROR");//设置日志等级为ERROR，避免日志增长导致的磁盘膨胀

        // 创建 Spark Streaming 上下文
        JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, new Duration(2000)); // 间隔两秒扑捉一次

        // 创建 Spark SQL 会话
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).getOrCreate();


        // 设置 Kafka 相关参数
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "10.0.0.105:9092,10.0.0.106:9092,10.0.0.107:9092");
        kafkaParams.put("key.deserializer", StringDeserializer.class);
        kafkaParams.put("value.deserializer", StringDeserializer.class);
        kafkaParams.put("auto.offset.reset", "earliest");
        // auto.offset.reset可指定参数有
        // latest：从分区的最新偏移量开始读取消息。
        // earliest：从分区的最早偏移量开始读取消息。
        // none：如果没有有效的偏移量，则抛出异常。
        kafkaParams.put("enable.auto.commit", true);  //采用自动提交offset 的模式
        kafkaParams.put("auto.commit.interval.ms",2000);//每隔离两秒提交一次commited-offset
        kafkaParams.put("group.id", "spark_kafka"); //消费组名称


        // 创建 Kafka stream
        Collection<String> topics = Collections.singletonList("spark_kafka"); // Kafka 主题名称
        JavaDStream<ConsumerRecord<String, String>> kafkaStream = KafkaUtils.createDirectStream(
                streamingContext,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams)  //订阅kafka
        );

        //定义数据结构
        StructType schema = new StructType()
                .add("key", DataTypes.LongType)
                .add("value", DataTypes.StringType);

        kafkaStream.foreachRDD((VoidFunction<JavaRDD<ConsumerRecord<String, String>>>) rdd -> {
            // 转换为 DataFrame
            Dataset<Row> df = sparkSession.createDataFrame(rdd.map(record -> {
                return RowFactory.create(record.offset(), record.value());  //将偏移量和value聚合
            }), schema);

            // 写入到 PostgreSQL
            df.write()
                    //选择写入数据库的模式
                    .mode(SaveMode.Append)//采用追加的写入模式
                    //协议
                    .format("jdbc")
                    //option 参数
                    .option("url", "jdbc:postgresql://localhost:5432/postgres") // PostgreSQL 连接 URL
                    //确定表名
                    .option("dbtable", "public.spark_kafka")//指定表名
                    .option("user", "postgres") // PostgreSQL 用户名
                    .option("password", "postgres") // PostgreSQL 密码
                    .save();
        });
        // 启动 Spark Streaming
        streamingContext.start();
        // 等待 Spark Streaming 应用程序终止
        streamingContext.awaitTermination();
    }
}

在执行代码前，向创建名为spark_kafka的topic

kafka-topics.sh --create --topic spark_kafka --bootstrap-server 10.0.0.105:9092,10.0.0.106:9092,10.0.0.107:9092

向spark_kafka 主题进行随机推数

kafka-producer-perf-test.sh --topic spark_kafka --thrghput 10 --num-records 10000 --record-size 100000 --producer-props bootstrap.servers=10.0.0.105:9092,10.0.0.106:9092,10.0.0.107:9092

运行过程中消费的offset会一直被提交到每一个分区
在这里插入图片描述

此时在数据库中查看，数据已经实时落地到库中
在这里插入图片描述

TCP

TCP环境下，实时监控日志的输出，可用于监控设备状态、环境变化等。当监测到异常情况时，可以实时发出警报。

package org.example;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.VoidFunction;
import org.apache.spark.sql.*;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructType;
import org.apache.spark.streaming.Duration;
import org.apache.spark.streaming.api.java.JavaDStream;
import org.apache.spark.streaming.api.java.JavaStreamingContext;
import org.apache.spark.streaming.kafka010.ConsumerStrategies;
import org.apache.spark.streaming.kafka010.KafkaUtils;
import org.apache.spark.streaming.kafka010.LocationStrategies;
import org.apache.kafka.common.serialization.StringDeserializer;

import java.util.*;

public class SparkStreamingKafka {
    public static void main(String[] args) throws InterruptedException {
        // 创建 Spark 配置
        SparkConf sparkConf = new SparkConf()
                .setAppName("spark_kafka") // 设置应用程序名称
                .setMaster("local[*]") // 设置 Spark master 为本地模式，[*]表示使用所有可用核心

                // 设置日志等级为ERROR，避免日志增长导致的磁盘膨胀
                .setExecutorEnv("setLogLevel", "ERROR");

        // 创建 Spark Streaming 上下文
        JavaStreamingContext streamingContext = new JavaStreamingContext(sparkConf, new Duration(2000)); // 间隔两秒扑捉一次

        // 创建 Spark SQL 会话
        SparkSession sparkSession = SparkSession.builder().config(sparkConf).getOrCreate();


        // 设置 Kafka 相关参数
        Map<String, Object> kafkaParams = new HashMap<>();
        kafkaParams.put("bootstrap.servers", "10.0.0.105:9092,10.0.0.106:9092,10.0.0.107:9092"); // Kafka 服务器地址
        kafkaParams.put("key.deserializer", StringDeserializer.class); // key 反序列化器类
        kafkaParams.put("value.deserializer", StringDeserializer.class); // value 反序列化器类
        kafkaParams.put("auto.offset.reset", "earliest"); // 从最早的偏移量开始消费消息
        kafkaParams.put("enable.auto.commit", true);  // 采用自动提交 offset 的模式
        kafkaParams.put("auto.commit.interval.ms", 2000); // 每隔两秒提交一次 committed-offset
        kafkaParams.put("group.id", "spark_kafka"); // 消费组名称


        // 创建 Kafka stream
        Collection<String> topics = Collections.singletonList("spark_kafka"); // Kafka 主题名称
        JavaDStream<ConsumerRecord<String, String>> kafkaStream = KafkaUtils.createDirectStream(
                streamingContext,
                LocationStrategies.PreferConsistent(),
                ConsumerStrategies.Subscribe(topics, kafkaParams)  // 订阅 Kafka
        );

        // 定义数据结构
        StructType schema = new StructType()
                .add("key", DataTypes.LongType)
                .add("value", DataTypes.StringType);

        kafkaStream.foreachRDD((VoidFunction<JavaRDD<ConsumerRecord<String, String>>>) rdd -> {
            // 转换为 DataFrame
            Dataset<Row> df = sparkSession.createDataFrame(rdd.map(record -> {
                return RowFactory.create(record.offset(), record.value());  // 将偏移量和 value 聚合
            }), schema);

            // 写入到 PostgreSQL
            df.write()
                    // 选择写入数据库的模式
                    .mode(SaveMode.Append) // 采用追加的写入模式
                    // 协议
                    .format("jdbc")
                    // option 参数
                    .option("url", "jdbc:postgresql://localhost:5432/postgres") // PostgreSQL 连接 URL
                    // 确定表名
                    .option("dbtable", "public.spark_kafka") // 指定表名
                    .option("user", "postgres") // PostgreSQL 用户名
                    .option("password", "postgres") // PostgreSQL 密码
                    .save();
        });
        // 启动 Spark Streaming
        streamingContext.start();
        // 等待 Spark Streaming 应用程序终止
        streamingContext.awaitTermination();
    }
}

在10.0.0.108 打开9999端口键入数值 ，使其被spark接收到并进行运算

nc -lk 9999

开启端口可以键入数值 此时会在IDEA的控制台显示其计算值
在这里插入图片描述

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：/a/422711.html

如若内容造成侵权/违法违规/事实不符，请联系我们进行投诉反馈qq邮箱809451989@qq.com，一经查实，立即删除！

相关文章

数通HCIE和云计算HCIE哪个好一点？

数通HCIE和云计算HCIE哪个好一点？

数通是网络的基础知识&#xff0c;也是入门人员必学的方向&#xff0c;相对也会简单些&#xff0c;学习数通&#xff0c;可以很好的学习其他的方向。数通的就业范围也比较广&#xff0c;运营商、企业、政府还是互联网公司&#xff0c;都需要大量的数通工程师来搭建和维护网络&a…
阅读更多...
科技企业如何做到FTP数据安全保护

科技企业如何做到FTP数据安全保护

在数字化浪潮的推动下&#xff0c;科技企业的数据已成为推动创新、提升效率、增强竞争力的核心资源。数据的重要性不言而喻&#xff0c;它不仅包含了客户信息、市场分析、产品设计等关键信息&#xff0c;更是企业宝贵的资产。然而&#xff0c;随着数据量的激增&#xff0c;数据…
阅读更多...
51单片机-（定时/计数器）

51单片机-（定时/计数器）

51单片机-&#xff08;定时/计数器&#xff09; 了解CPU时序、特殊功能寄存器和定时/计数器工作原理&#xff0c;以定时器0实现每次间隔一秒亮灯一秒的实验为例理解定时/计数器的编程实现。 1.CPU时序 1.1.四个周期 振荡周期&#xff1a;为单片机提供定时信号的振荡源的周期…
阅读更多...
【论文笔记】Mamba：挑战Transformer地位的新架构

【论文笔记】Mamba：挑战Transformer地位的新架构

Mamba Mamba: Linear-Time Sequence Modeling with Selective State Spaces 代码地址 摘要 背景 基础模型&#xff0c;它现在驱动着深度学习中大多数令人振奋的应用&#xff0c;几乎都是基于Transformer架构及其核心注意力模块。 存在的问题 许多subquadratic-time架构如…
阅读更多...
langchain学习笔记（十）

langchain学习笔记（十）

Bind runtime args | &#x1f99c;️&#x1f517; Langchain 1、有时&#xff0c;我们希望使用常量参数调用Runnable序列中的Runnable&#xff0c;这些参数不是序列中前一个Runnable的输出的一部分&#xff0c;也不是用户的输入&#xff0c;这时可以用Runnable.bind() from …
阅读更多...
【学习心得】浏览器开发者工具中出现的VM开头的JS文件是什么？

【学习心得】浏览器开发者工具中出现的VM开头的JS文件是什么？

一、现象描述 在Chrome的开发者工具中&#xff0c;你可能会看到一些以“VM”开头的JavaScript文件&#xff08;如“VM111.js”&#xff09;。 二、VM文件到底是什么&#xff1f; “VM”表示的是Virtual Machine&#xff08;虚拟机&#xff09;&#xff0c;这些文件通常表示由浏…
阅读更多...
C++：菱形继承问题

C++：菱形继承问题

目录 1、什么是菱形继承 2、虚拟继承 3、一些常见问题 1. 什么是菱形继承&#xff1f;菱形继承的问题是什么&#xff1f; 2. 什么是菱形虚拟继承&#xff1f;如何解决数据冗余和二义性的 3. 继承和组合的区别&#xff1f;什么时候用继承&#xff1f;什么时候用组合&#…
阅读更多...
wpsexcel蛋疼的数字列转文本

wpsexcel蛋疼的数字列转文本

在excel处理时&#xff0c;有时某一列会以数字形式展示&#xff0c;后续处理非常麻烦&#xff0c;希望转为字符&#xff0c;参考以下方法 选定整列点击"数据"->“分列”->一直下一步直到选择文本
阅读更多...
Java求职技能清单（2024版）

Java求职技能清单（2024版）

一、Java基础扎实&#xff08;反射、集合、IO、NIO、多线程、设计模式、通信协议等基础技术&#xff09; &#xff08;一&#xff09;Java &#xff08;二&#xff09;网络IO &#xff08;三&#xff09;NIO模型 &#xff08;…
阅读更多...
在golang中使用protoc

在golang中使用protoc

【Golang】proto生成go的相关文件 推荐个人主页&#xff1a;席万里的个人空间 文章目录 【Golang】proto生成go的相关文件1、查看proto的版本号2、安装protoc-gen-go和protoc-gen-go-grpc3、生成protobuff以及grpc的文件 1、查看proto的版本号 protoc --version2、安装protoc-…
阅读更多...
备战蓝桥杯---线段树基础2

备战蓝桥杯---线段树基础2

今天我们把线段树的另一个模板看一下&#xff1a; 在这里&#xff0c;我们注意到乘的操作&#xff0c;因此我们用两个懒标记来分别表示加和乘&#xff0c;这时我们面临了一个问题&#xff0c;就是当我们把标记往下传时&#xff0c;它的儿子怎么知道是先乘还是先加&#xff1f; …
阅读更多...
Doris实战——拈花云科的数据中台实践

Doris实战——拈花云科的数据中台实践

前言 拈花云科 NearFar X Lab 团队调研并引进 Doris 作为新架构下的数据仓库选型方案。本文主要介绍了拈花云科数据中台架构从 1.0 到 2.0 的演变过程&#xff0c;以及 Doris 在交付型项目和 SaaS 产品中的应用实践。 一、业务背景 拈花云科的服务对象主要是国内各个景区、景点…
阅读更多...
React-router的创建和第一个组件

React-router的创建和第一个组件

需要先学react框架 首先&#xff1a;找到一个文件夹&#xff0c;在文件夹出打开cmd窗口&#xff0c;输入如下图的口令 npx create-react-app demo 然后等待安装 安装完成 接下来进入创建的demo实例 cd demo 然后可以用如下方式打开vscode code . 注意&#xff1a;不要忽略点号与…
阅读更多...
【重温设计模式】享元模式及其Java示例

【重温设计模式】享元模式及其Java示例

享元模式的介绍 在编程世界中&#xff0c;我们常常面临着如何有效管理系统资源的挑战。这就好比我们在生活中&#xff0c;面对有限的物质资源&#xff0c;如何做到既满足需求又节约使用&#xff0c;是一门艺术。在设计模式中&#xff0c;有一种模式&#xff0c;恰如其分地解决…
阅读更多...
VR转接器：破解虚拟与现实边界的革命性设备

VR转接器：破解虚拟与现实边界的革命性设备

VR转接器&#xff0c;这一革命性的设备&#xff0c;为虚拟现实体验带来了前所未有的自由度。它巧妙地连接了虚拟与现实&#xff0c;使得用户在享受VR眼镜带来的奇幻世界的同时&#xff0c;也能自由地在现实世界中活动。这一设计的诞生&#xff0c;不仅解决了VR眼镜续航的瓶颈问…
阅读更多...
MySQL进阶之（三）InnoDB数据存储结构之数据页结构

MySQL进阶之（三）InnoDB数据存储结构之数据页结构

三、InnoDB数据存储结构之数据页结构 3.1 数据库的存储结构3.1.1 MySQL 数据存储目录3.1.2 页的引入3.1.3 页的概述3.1.4 页的上层结构 3.2 数据页结构3.2.1 文件头和文件尾01、File Header&#xff08;文件头部&#xff09;02、File Trailer&#xff08;文件尾部&#xff09; …
阅读更多...
比小鹏、问界都贵，谁给了理想MEGA勇气？

比小鹏、问界都贵，谁给了理想MEGA勇气？

“规模小的时候&#xff0c;一号位善于解题。规模大的时候&#xff0c;一号位要善于出题。” 前不久&#xff0c;理想汽车CEO李想在微博上如此评价一家公司中&#xff0c;老板应该怎么做。 现在&#xff0c;成立近9年的理想汽车做出了一个“违背祖宗”的决定——大举进军纯电…
阅读更多...
陶瓷工业5G智能制造工厂数字孪生可视化平台，推进行业数字化转型

陶瓷工业5G智能制造工厂数字孪生可视化平台，推进行业数字化转型

陶瓷工业5G智能制造工厂数字孪生可视化平台&#xff0c;推进行业数字化转型。在陶瓷工业领域&#xff0c;5G智能制造工厂数字孪生可视化平台的应用正在改变着行业的传统生产模式&#xff0c;推动着数字化转型的进程。本文将围绕这一主题展开探讨&#xff0c;分析数字孪生可视化…
阅读更多...
挑战30天学完Python：Day25 pandas

挑战30天学完Python：Day25 pandas

&#x1f389; 本系列为Python基础学习&#xff0c;原稿来源于 30-Days-Of-Python 英文项目&#xff0c;大奇主要是对其本地化翻译、逐条验证和补充&#xff0c;想通过30天完成正儿八经的系统化实践。此系列适合零基础同学&#xff0c;或仅了解Python一点知识&#xff0c;但又没…
阅读更多...
智能家居控制系统（51单片机）

智能家居控制系统（51单片机）

smart_home_control_system 51单片机课设&#xff0c;智能家居控制系统 使用及转载请标明出处&#xff08;最好点个赞及star哈哈&#xff09; Github地址&#xff0c;带有PPT及流程图 Gitee码云地址&#xff0c;带有PPT及流程图 ​ 以STC89C52为主控芯片&#xff0c;以矩阵键…
阅读更多...
最新文章

Copyright @ 2022~2023