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

Spark-Streaming+Kafka+mysql实战示例

文章目录

  • 前言
  • 一、简介
    • 1. Spark-Streaming简介
    • 2. Kafka简介
  • 二、实战演练
    • 1. MySQL数据库部分
    • 2. 导入依赖
    • 3. 编写实体类代码
    • 4. 编写kafka主题管理代码
    • 5. 编写kafka生产者代码
    • 6. 编写Spark-Streaming代码
    • 7. 查看数据库
    • 8. 代码下载
  • 总结


前言

本文将介绍一个使用Spark Streaming和Kafka进行实时数据处理的示例。通过该示例,您将了解到如何使用Spark Streaming和Kafka处理实时数据流,以及如何将处理后的数据保存到MySQL数据库中。示例涵盖了从环境搭建到代码实现的全过程,帮助您快速上手实时数据处理的开发。

zookeeper安装教程:zookeeper安装与配置:使用shell脚本在centos上进行zookeeper自动化下载安装配置(集群搭建版)
kafka安装教程:Kafka安装与配置-shell脚本一键安装配置(集群版)


一、简介

1. Spark-Streaming简介

Spark Streaming是Apache Spark的一个组件,用于实时流数据处理。它提供了高级别的API,可以使用类似于批处理的方式处理实时数据流。Spark Streaming可以与各种消息队列系统集成,包括Kafka、RabbitMQ等。

2. Kafka简介

Kafka是一个分布式流处理平台,具有高吞吐量、可扩展性和可靠性。它提供了一种可持久化、分布式、分区的日志服务,用于处理实时数据流。Kafka使用发布-订阅模型,消息被发布到一个或多个主题,然后由订阅该主题的消费者进行消费。


二、实战演练

开始之前先启动zookeeper集群和kafka集群。

1. MySQL数据库部分

这部分代码用于创建MySQL数据库和数据表,以及将从Kafka获取的数据保存到数据库中。

create database kafkademo;

创建数据表:

CREATE TABLE kafka_tb
(`txid`      varchar(255) PRIMARY KEY,`version`   varchar(255),`connector` varchar(255),`name`      varchar(255),`ts_ms`     varchar(255),`snapshot`  varchar(255),`db`        varchar(255),`sequence`  varchar(255),`schema`    varchar(255),`table`     varchar(255),`lsn`       varchar(255),`xmin`      varchar(255)
);

2. 导入依赖

这部分代码是Maven的依赖配置,用于引入所需的Spark、Kafka和MySQL相关的库。

<dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.26</version>
</dependency>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-core_2.11</artifactId><version>2.4.0</version>
</dependency>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-sql_2.11</artifactId><version>2.4.0</version>
</dependency>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming_2.11</artifactId><version>2.4.0</version>
</dependency>
<dependency><groupId>org.apache.spark</groupId><artifactId>spark-streaming-kafka-0-10_2.11</artifactId><version>2.4.0</version>
</dependency>
<dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.8.0</version>
</dependency>
<dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.26</version><scope>compile</scope>
</dependency>

3. 编写实体类代码

这部分代码定义了一个Java类EntityMessage,用于将从Kafka获取的JSON数据转换为Java对象。

package com.zcs;import lombok.Data;import java.io.Serializable;/*** @author zcs2312* @date 2023/12/12 20:49:47* @product_name IntelliJ IDEA* @project_name spark-kafka*/
@Data
public class EntityMessage implements Serializable {private String op;private String ts_ms;private String transaction;private DataItem dataItem;@Datapublic static class DataItem {private String version;private String connector;private String name;private String ts_ms;private String snapshot;private String db;private String[] sequence;private String schema;private String table;private String txId;private String lsn;private String xmin;}
}

4. 编写kafka主题管理代码

这部分代码用于创建、删除和修改Kafka主题的一些操作。

package com.zcs;import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.config.ConfigResource;import java.util.*;
import java.util.concurrent.ExecutionException;/*** @author zcs2312* @date 2023/12/12 20:51:34* @product_name IntelliJ IDEA* @project_name spark-kafka*/
public class KafkaTopicManager {

相关文章:

Spark-Streaming+Kafka+mysql实战示例

文章目录 前言一、简介1. Spark-Streaming简介2. Kafka简介二、实战演练1. MySQL数据库部分2. 导入依赖3. 编写实体类代码4. 编写kafka主题管理代码5. 编写kafka生产者代码6. 编写Spark-Streaming代码7. 查看数据库8. 代码下载总结前言 本文将介绍一个使用Spark Streaming和Ka…...

C++改写为C

stm使用中&#xff0c;经常能见到CPP的示例&#xff0c;这些是给arduino&#xff0c;esp32用的&#xff0c;stm32 也支持cpp但是你就想用c怎么办呢&#xff0c;比如我在新手的时候&#xff1a;&#xff1a; 这个双冒号就难住了英雄好汉 比如这是个cpp的 如果类不多的情况下 改写…...

抖去推--短视频剪辑、矩阵无人直播saas营销工具一站式开发

抖去推是一款短视频剪辑和矩阵无人直播SAAS营销工具一站式开发平台。它提供了以下功能和特点&#xff1a; 1. 短视频剪辑&#xff1a;抖去推提供了一系列的剪辑工具&#xff0c;包括自动剪辑、特效制作、配音配乐等&#xff0c;可以帮助用户轻松制作出高质量的短视频。 2. 矩阵…...

HBase 详细图文介绍

目录 一、HBase 定义 二、HBase 数据模型 2.1 HBase 逻辑结构 2.2 HBase 物理存储结构 ​2.3 数据模型 2.3.1 Name Space 2.3.2 Table 2.3.3 Row 2.3.4 Column 2.3.5 Time Stamp 2.3.6 Cell 三、HBase 基本架构 架构角色 3.1 Master 3.2 Region Server 3.3 Zo…...

Hanlp自然语言处理如何再Spring Boot中使用

一、HanLP HanLP (Hankcs NLP) 是一个自然语言处理工具包&#xff0c;具有功能强大、性能高效、易于使用的特点。HanLP 主要支持中文文本处理&#xff0c;包括分词、词性标注、命名实体识别、依存句法分析、关键词提取、文本分类、情感分析等多种功能。 HanLP 可以在 Java、Py…...

MySQL 是什么?

MySQL官方网站&#xff08;http://www.mysql.com/&#xff09;提供关于MySQL软件的最新信息。 MySQL是一个数据库管理系统。 数据库是一种结构化的数据集合。它可以是从简单的购物清单到图片库&#xff0c;再到企业网络中的大量信息等任何形式。要添加、访问和处理存储在计算…...

yarn link使用(npm link)

使用场景 前端开发中&#xff0c;两个项目相互依赖时&#xff0c;使用yarn link(npm link)链接 例如&#xff1a;A项目依赖于本司自己的UI库B&#xff0c;当我们修改了UI库B中的某些代码时&#xff0c;需本地验证后再发布到私服&#xff0c;此时A项目与UI项目B通过yarn link连…...

Docker容器讲解

Docker是一个开源的容器化平台&#xff0c;可以用来在轻量级容器中打包、部署和运行应用程序。Docker的基本概念包括容器、镜像、仓库和服务。 容器是一个独立运行的应用程序包&#xff0c;包括应用程序及其依赖项、运行时环境和配置等。容器相互隔离&#xff0c;可以在不同的…...

three.js模拟太阳系

地球的旋转轨迹目前设置为了圆形&#xff0c;效果&#xff1a; <template><div><el-container><el-main><div class"box-card-left"><div id"threejs" style"border: 1px solid red"></div><div c…...

WPF仿网易云搭建笔记(1):项目搭建

文章目录 前言项目地址动态样式组合样式批量样式覆盖Prism新建UserControler修改Material Design 笔刷收放列表可以滚动的StackPanel列表点击展开或折叠 实现效果 前言 今天接着继续细化代码&#xff0c;把整体框架写出来 项目地址 WPF仿网易云 Gitee仓库 动态样式 【WPF】C#…...

DDOS 攻击是什么?有哪些常见的DDOS攻击?

DDOS简介 DDOS又称为分布式拒绝服务&#xff0c;全称是Distributed Denial of Service。DDOS本是利用合理的请求造成资源过载&#xff0c;导致服务不可用&#xff0c;从而造成服务器拒绝正常流量服务。就如酒店里的房间是有固定的数量的&#xff0c;比如一个酒店有50个房间&am…...

未来应用从何而来:认知力延伸、边界突破、回归云与产业

文 | 智能相对论 作者 | 沈浪 或许&#xff0c;谁也没想到未来应用来的如此之快&#xff0c;现如今传统应用从开发到体验&#xff0c;已经进入了一个前所未有的颠覆性改革阶段。 不久前&#xff0c;美国人工智能公司OpenAI举办开发者大会。在现场&#xff0c;公司创始人Sam …...

vue零基础

vue 与其他框架的对比 框架设计模式数据绑定灵活度文件模式复杂性学习曲线生态VueMVVM双向灵活单文件小缓完善ReactMVC单向较灵活all in js大陡丰富AngularMVC双向固定多文件较大较陡&#xff08;Typescript&#xff09;独立 更多对比细节&#xff1a;vue 官网&#xff1a;ht…...

html中一个div中平均一行分配四个盒子,可展开与收起所有的盒子

html中一个div中平均一行分配四个盒子&#xff0c;可展开与收起所有的盒子 1.截图显示部分 2.代码展示部分 <!DOCTYPE html> <html lang"en"><head><meta charset"UTF-8"><meta name"viewport" content"wid…...

Python虚拟环境指南:告别依赖地狱

一、背景 在SAAS&#xff08;软件即服务&#xff09;平台中&#xff0c;用户使用自行定制的Python脚本已经成为司空见惯的做法&#xff0c;然而&#xff0c;由于不同用户对Python三方库的需求各不相同&#xff0c;而底层服务器一般只安装了一个Python版本。举例来说&#xff0…...

【Jeecg Boot 3 - 第二天】第2节 前后端docker部署云服务器

更新完成&#xff0c;点击下面章节进入 一、后端部署 1.1、后端 docker-compose 部署 JEECGBOOT 1.2、jar 包和 lib 依赖分离&#xff0c;部署包缩小100倍 二、前端部署 2.1、nginx 部署 JEECGBOOT VUE3 2.2、开启Nginx压缩&#xff0c;解决前端访问慢问题...

2020年第九届数学建模国际赛小美赛A题自由泳解题全过程文档及程序

2020年第九届数学建模国际赛小美赛 A题 自由泳 原题再现&#xff1a; 在所有常见的游泳泳姿中&#xff0c;哪一种最快&#xff1f;哪个冲程推力最大&#xff1f;在自由泳项目中&#xff0c;游泳者可以选择他们的泳姿&#xff0c;他们通常选择前面的爬行。然而&#xff0c;游泳…...

双端队列和优先级队列

文章目录 前言dequedeque底层设计迭代器设计 priority仿函数数组中的第k个最大元素优先级队列模拟实现pushpop调整仿函数存储自定义类型 前言 今天要介绍比较特殊的结构&#xff0c;双端队列。 还有一个适配器&#xff0c;优先级队列。 deque 栈的默认容器用了一个deque的东西…...

c#读取CSV文件跟Excel导入成DataTble

1.读取CSV文件 /// <summary>/// 读取CSV文件/// </summary>/// <param name"fileName">文件路径</param>public static DataTable ReadCSV(string fileName){DataTable dt new DataTable();FileStream fs new FileStream(fileName, FileM…...

Python编程技巧 – 单字符函数

Python编程技巧 – 单字符函数 Python Programming Skills – Single Character Function By JacksonML 0. 前言 Python有其内建(built-in)的一系列函数&#xff0c;其中&#xff0c;有两个函数为长度为一的字符设计。这样的函数是单字符函数&#xff0c;尽管它们操作的对象…...

设计模式和设计原则回顾

设计模式和设计原则回顾 23种设计模式是设计原则的完美体现,设计原则设计原则是设计模式的理论基石, 设计模式 在经典的设计模式分类中(如《设计模式:可复用面向对象软件的基础》一书中),总共有23种设计模式,分为三大类: 一、创建型模式(5种) 1. 单例模式(Sing…...

macOS多出来了:Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用

文章目录 问题现象问题原因解决办法 问题现象 macOS启动台&#xff08;Launchpad&#xff09;多出来了&#xff1a;Google云端硬盘、YouTube、表格、幻灯片、Gmail、Google文档等应用。 问题原因 很明显&#xff0c;都是Google家的办公全家桶。这些应用并不是通过独立安装的…...

【ROS】Nav2源码之nav2_behavior_tree-行为树节点列表

1、行为树节点分类 在 Nav2(Navigation2)的行为树框架中,行为树节点插件按照功能分为 Action(动作节点)、Condition(条件节点)、Control(控制节点) 和 Decorator(装饰节点) 四类。 1.1 动作节点 Action 执行具体的机器人操作或任务,直接与硬件、传感器或外部系统…...

第一篇:Agent2Agent (A2A) 协议——协作式人工智能的黎明

AI 领域的快速发展正在催生一个新时代&#xff0c;智能代理&#xff08;agents&#xff09;不再是孤立的个体&#xff0c;而是能够像一个数字团队一样协作。然而&#xff0c;当前 AI 生态系统的碎片化阻碍了这一愿景的实现&#xff0c;导致了“AI 巴别塔问题”——不同代理之间…...

反射获取方法和属性

Java反射获取方法 在Java中&#xff0c;反射&#xff08;Reflection&#xff09;是一种强大的机制&#xff0c;允许程序在运行时访问和操作类的内部属性和方法。通过反射&#xff0c;可以动态地创建对象、调用方法、改变属性值&#xff0c;这在很多Java框架中如Spring和Hiberna…...

EtherNet/IP转DeviceNet协议网关详解

一&#xff0c;设备主要功能 疆鸿智能JH-DVN-EIP本产品是自主研发的一款EtherNet/IP从站功能的通讯网关。该产品主要功能是连接DeviceNet总线和EtherNet/IP网络&#xff0c;本网关连接到EtherNet/IP总线中做为从站使用&#xff0c;连接到DeviceNet总线中做为从站使用。 在自动…...

鱼香ros docker配置镜像报错:https://registry-1.docker.io/v2/

使用鱼香ros一件安装docker时的https://registry-1.docker.io/v2/问题 一键安装指令 wget http://fishros.com/install -O fishros && . fishros出现问题&#xff1a;docker pull 失败 网络不同&#xff0c;需要使用镜像源 按照如下步骤操作 sudo vi /etc/docker/dae…...

深入解析C++中的extern关键字:跨文件共享变量与函数的终极指南

&#x1f680; C extern 关键字深度解析&#xff1a;跨文件编程的终极指南 &#x1f4c5; 更新时间&#xff1a;2025年6月5日 &#x1f3f7;️ 标签&#xff1a;C | extern关键字 | 多文件编程 | 链接与声明 | 现代C 文章目录 前言&#x1f525;一、extern 是什么&#xff1f;&…...

今日学习:Spring线程池|并发修改异常|链路丢失|登录续期|VIP过期策略|数值类缓存

文章目录 优雅版线程池ThreadPoolTaskExecutor和ThreadPoolTaskExecutor的装饰器并发修改异常并发修改异常简介实现机制设计原因及意义 使用线程池造成的链路丢失问题线程池导致的链路丢失问题发生原因 常见解决方法更好的解决方法设计精妙之处 登录续期登录续期常见实现方式特…...

GC1808高性能24位立体声音频ADC芯片解析

1. 芯片概述 GC1808是一款24位立体声音频模数转换器&#xff08;ADC&#xff09;&#xff0c;支持8kHz~96kHz采样率&#xff0c;集成Δ-Σ调制器、数字抗混叠滤波器和高通滤波器&#xff0c;适用于高保真音频采集场景。 2. 核心特性 高精度&#xff1a;24位分辨率&#xff0c…...