使用Flink实现MySQL到Kafka的数据流转换

使用Flink实现MySQL到Kafka的数据流转换

本篇博客将介绍如何使用Flink将数据从MySQL数据库实时传输到Kafka，这是一个常见的用例，适用于需要实时数据connector的场景。
在这里插入图片描述

环境准备

在开始之前，确保你的环境中已经安装了以下软件：
Apache Flink 准备相关pom依赖

<?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.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>org.example</groupId><artifactId>EastMoney</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.flink</groupId><artifactId>flink-clients_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-scala-bridge_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-scala_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>1.14.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>1.14.0</version></dependency><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>8.0.25</version></dependency></dependencies></project>

MySQL数据库，初始化mysql表

CREATE TABLE `t_stock_code_price` (`id` bigint NOT NULL AUTO_INCREMENT,`code` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票代码',`name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票名称',`close` double DEFAULT NULL COMMENT '最新价',`change_percent` double DEFAULT NULL COMMENT '涨跌幅',`change` double DEFAULT NULL COMMENT '涨跌额',`volume` double DEFAULT NULL COMMENT '成交量（手）',`amount` double DEFAULT NULL COMMENT '成交额',`amplitude` double DEFAULT NULL COMMENT '振幅',`turnover_rate` double DEFAULT NULL COMMENT '换手率',`peration` double DEFAULT NULL COMMENT '市盈率',`volume_rate` double DEFAULT NULL COMMENT '量比',`hign` double DEFAULT NULL COMMENT '最高',`low` double DEFAULT NULL COMMENT '最低',`open` double DEFAULT NULL COMMENT '今开',`previous_close` double DEFAULT NULL COMMENT '昨收',`pb` double DEFAULT NULL COMMENT '市净率',`create_time` varchar(64) NOT NULL COMMENT '写入时间',PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=5605 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

Kafka消息队列

1. 启动zookeeperzkServer start
2. 启动kafka服务kafka-server-start /opt/homebrew/etc/kafka/server.properties
3. 创建topickafka-topics --create --bootstrap-server 127.0.0.1:9092 --replication-factor 1 --partitions 1 --topic east_money
4. 消费数据kafka-console-consumer --bootstrap-server 127.0.0.1:9092 --topic east_money --from-beginning

步骤解释

获取流执行环境：首先，我们通过StreamExecutionEnvironment.getExecutionEnvironment获取Flink的流执行环境，并设置其运行模式为流处理模式。

创建流表环境：接着，我们通过StreamTableEnvironment.create创建一个流表环境，这个环境允许我们使用SQL语句来操作数据流。

val senv = StreamExecutionEnvironment.getExecutionEnvironment.setRuntimeMode(RuntimeExecutionMode.STREAMING)val tEnv = StreamTableEnvironment.create(senv)

定义MySQL数据源表：我们使用一个SQL语句创建了一个临时表t_stock_code_price，这个表代表了我们要从MySQL读取的数据结构和连接信息。

val source_table ="""|CREATE TEMPORARY TABLE t_stock_code_price (|  id BIGINT NOT NULL,|  code STRING NOT NULL,|  name STRING NOT NULL,|  `close` DOUBLE,|  change_percent DOUBLE,|  change DOUBLE,|  volume DOUBLE,|  amount DOUBLE,|  amplitude DOUBLE,|  turnover_rate DOUBLE,|  peration DOUBLE,|  volume_rate DOUBLE,|  hign DOUBLE,|  low DOUBLE,|  `open` DOUBLE,|  previous_close DOUBLE,|  pb DOUBLE,|  create_time STRING NOT NULL,|  PRIMARY KEY (id) NOT ENFORCED|) WITH (|   'connector' = 'jdbc',|   'url' = 'jdbc:mysql://localhost:3306/mydb',|   'driver' = 'com.mysql.cj.jdbc.Driver',|   'table-name' = 't_stock_code_price',|   'username' = 'root',|   'password' = '12345678'|)|""".stripMargintEnv.executeSql(source_table)

定义Kafka目标表：然后，我们定义了一个Kafka表re_stock_code_price_kafka，指定了Kafka的连接参数和表结构。

tEnv.executeSql("CREATE TABLE re_stock_code_price_kafka (" +"`id` BIGINT," +"`code` STRING," +"`name` STRING," +"`close` DOUBLE," +"`change_percent` DOUBLE," +"`change` DOUBLE," +"`volume` DOUBLE," +"`amount` DOUBLE," +"`amplitude` DOUBLE," +"`turnover_rate` DOUBLE," +"`operation` DOUBLE," +"`volume_rate` DOUBLE," +"`high` DOUBLE," +"`low` DOUBLE," +"`open` DOUBLE," +"`previous_close` DOUBLE," +"`pb` DOUBLE," +"`create_time` STRING," +"rise int"+") WITH (" +"'connector' = 'kafka'," +"'topic' = 'east_money'," +"'properties.bootstrap.servers' = '127.0.0.1:9092'," +"'properties.group.id' = 'mysql2kafka'," +"'scan.startup.mode' = 'earliest-offset'," +"'format' = 'csv'," +"'csv.field-delimiter' = ','" +")")

数据转换和写入：最后，我们执行了一个插入操作，将从MySQL读取的数据转换（这里通过case when语句添加了一个新字段rise）并写入到Kafka中。这个可以实现任何的sql etl 来满足我们的需求。

    tEnv.executeSql("insert into re_stock_code_price_kafka select *,case when change_percent>0 then 1 else 0 end as rise from t_stock_code_price")

全部代码

package org.eastimport org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironmentobject Mysql2Kafka {def main(args: Array[String]): Unit = {val senv = StreamExecutionEnvironment.getExecutionEnvironment.setRuntimeMode(RuntimeExecutionMode.STREAMING)val tEnv = StreamTableEnvironment.create(senv)val source_table ="""|CREATE TEMPORARY TABLE t_stock_code_price (|  id BIGINT NOT NULL,|  code STRING NOT NULL,|  name STRING NOT NULL,|  `close` DOUBLE,|  change_percent DOUBLE,|  change DOUBLE,|  volume DOUBLE,|  amount DOUBLE,|  amplitude DOUBLE,|  turnover_rate DOUBLE,|  peration DOUBLE,|  volume_rate DOUBLE,|  hign DOUBLE,|  low DOUBLE,|  `open` DOUBLE,|  previous_close DOUBLE,|  pb DOUBLE,|  create_time STRING NOT NULL,|  PRIMARY KEY (id) NOT ENFORCED|) WITH (|   'connector' = 'jdbc',|   'url' = 'jdbc:mysql://localhost:3306/mydb',|   'driver' = 'com.mysql.cj.jdbc.Driver',|   'table-name' = 't_stock_code_price',|   'username' = 'root',|   'password' = '12345678'|)|""".stripMargintEnv.executeSql(source_table)val result = tEnv.executeSql("select * from t_stock_code_price")result.print()tEnv.executeSql("CREATE TABLE re_stock_code_price_kafka (" +"`id` BIGINT," +"`code` STRING," +"`name` STRING," +"`close` DOUBLE," +"`change_percent` DOUBLE," +"`change` DOUBLE," +"`volume` DOUBLE," +"`amount` DOUBLE," +"`amplitude` DOUBLE," +"`turnover_rate` DOUBLE," +"`operation` DOUBLE," +"`volume_rate` DOUBLE," +"`high` DOUBLE," +"`low` DOUBLE," +"`open` DOUBLE," +"`previous_close` DOUBLE," +"`pb` DOUBLE," +"`create_time` STRING," +"rise int"+") WITH (" +"'connector' = 'kafka'," +"'topic' = 'east_money'," +"'properties.bootstrap.servers' = '127.0.0.1:9092'," +"'properties.group.id' = 'mysql2kafka'," +"'scan.startup.mode' = 'earliest-offset'," +"'format' = 'csv'," +"'csv.field-delimiter' = ','" +")")tEnv.executeSql("insert into re_stock_code_price_kafka select *,case when change_percent>0 then 1 else 0 end as rise from t_stock_code_price")}
}

如有遇到问题可以找小编沟通交流哦。另外小编帮忙辅导大课作业，学生毕设等。不限于python，java，大数据，模型训练等。
在这里插入图片描述

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

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈email:809451989@qq.com，一经查实，立即删除！

相关文章

ML-Decoder: Scalable and Versatile Classification Head

ML-Decoder: Scalable and Versatile Classification Head

1、引言 论文链接&#xff1a;https://openaccess.thecvf.com/content/WACV2023/papers/Ridnik_ML-Decoder_Scalable_and_Versatile_Classification_Head_WACV_2023_paper.pdf 因为 transformer 解码器分类头[1] 在少类别多标签分类数据集上表现得很好&#xff0c;但由于其查询…
阅读更多...
公平抽签（蓝桥杯）

公平抽签（蓝桥杯）

文章目录 公平抽签题目描述回溯算法 公平抽签 题目描述 小A的学校&#xff0c;蓝桥杯的参赛名额非常有限&#xff0c;只有 m 个名额&#xff0c;但是共有 n 个人报名。 作为老师非常苦恼&#xff0c;他不知道该让谁去&#xff0c;他在寻求一个绝对公平的方式。 于是他准备让…
阅读更多...
PHP的定时任务框架的taskPHP3.0学习记录2（环境要求、配置Redis、crontab执行时间语法、命令操作以及Screen全屏窗口管理器）

PHP的定时任务框架的taskPHP3.0学习记录2（环境要求、配置Redis、crontab执行时间语法、命令操作以及Screen全屏窗口管理器）

环境要求 php版本> 5.5开启socket扩展开启pdo扩展开启shmop扩展 echo <pre>; echo --; $requiredVersion 5.6.0; $currentVersion phpversion(); if (version_compare($currentVersion, $requiredVersion, >)) {echo "1.PHP版本满足要求&#xff0c;当前版…
阅读更多...
c语言：vs2022写一个一元二次方程（包含虚根）

c语言：vs2022写一个一元二次方程（包含虚根）

求一元二次方程 的根&#xff0c;通过键盘输入a、b、c&#xff0c;根据△的值输出对应x1和x2的值(保留一位小数)(用if语句完成)。 //一元二次方程的实现 #include <stdio.h> #include <math.h> #include <stdlib.h> int main() {double a, b, c, delta, x1…
阅读更多...
数据结构 - 算法效率|时间复杂度|空间复杂度

数据结构 - 算法效率|时间复杂度|空间复杂度

目录 1.算法效率 2.时间复杂度 2.1定义 2.2大O渐近表示法 2.3常见时间复杂度计算举例 3.空间复杂度 3.1定义 3.2常见空间复杂度计算举例 1.算法效率 算法的效率常用算法复杂度来衡量&#xff0c;算法复杂度描述了算法在输入数据规模变化时&#xff0c;其运行时间和空间…
阅读更多...
opejdk11 java 启动流程 java main方法怎么被jvm执行

opejdk11 java 启动流程 java main方法怎么被jvm执行

java启动过程 java main方法怎么被jvm执行 java main方法是怎么被jvm调用的 1、jvm main入口 2、执行JLI_Launch方法 3、执行JVMInit方法 4、执行ContinueInNewThread方法 5、执行CallJavaMainInNewThread方法 6、创建线程执行ThreadJavaMain方法 7、执行ThreadJavaMain方法…
阅读更多...
Last-Modified：HTTP缓存控制机制解析

Last-Modified：HTTP缓存控制机制解析

&#x1f90d; 前端开发工程师、技术日更博主、已过CET6 &#x1f368; 阿珊和她的猫_CSDN博客专家、23年度博客之星前端领域TOP1 &#x1f560; 牛客高级专题作者、打造专栏《前端面试必备》 、《2024面试高频手撕题》 &#x1f35a; 蓝桥云课签约作者、上架课程《Vue.js 和 E…
阅读更多...
docker配置github仓库ghcr国内镜像加速

docker配置github仓库ghcr国内镜像加速

文章目录 说明ghcr.io简介配置镜像命令地址命令行方式1panel面板方式方式一&#xff1a;配置镜像加速&#xff0c;命令行拉取方式二&#xff1a;配置镜像仓库&#xff0c;可视化拉取 说明 由于使用的容器需要从github下载镜像&#xff0c;服务器在国外下载速度很慢&#xff0c…
阅读更多...
26. UE5 RPG同步面板属性（二）

26. UE5 RPG同步面板属性（二）

在上一篇&#xff0c;我们解析了UI属性面板的实现步骤&#xff1a; 首先我们需要通过c去实现创建GameplayTag&#xff0c;这样可以在c和UE里同时获取到Tag创建一个DataAsset类&#xff0c;用于设置tag对应的属性和显示内容创建AttributeMenuWidgetController实现对应逻辑 并且…
阅读更多...
理解游戏服务器架构-部署架构

理解游戏服务器架构-部署架构

目录 前言 我所理解的服务器架构 什么是否部署架构 部署架构的职责 进程业务职责 网络链接及通讯方式 与客户端的连接方式 服务器之间连接关系 数据落地以及一致性 数据库的选择 数据访问三级缓存 数据分片 读写分离 分布式数据处理 负载均衡 热更新 配置更新 …
阅读更多...
html第二次作业

html第二次作业

骨架 <!DOCTYPE html> <html lang"en"> <head><meta charset"UTF-8"><meta http-equiv"X-UA-Compatible" content"IEedge"><meta name"viewport" content"widthdevice-width, initi…
阅读更多...
vscode初始化node项目

vscode初始化node项目

首先需要安装node环境&#xff0c;推荐直接使用nvm 安装node&#xff0c;方便切换node版本 1.npm init 初始化node项目 在命令行输入npm init指令 根据指令创建完成后会在当前目录下生成一个package.json文件&#xff0c;记住运行npm init执行的目录必须是一个空目录 2.创建…
阅读更多...
蓝桥杯算法题-发现环

蓝桥杯算法题-发现环

问题描述 　　小明的实验室有N台电脑&#xff0c;编号1~N。原本这N台电脑之间有N-1条数据链接相连&#xff0c;恰好构成一个树形网络。在树形网络上&#xff0c;任意两台电脑之间有唯一的路径相连。 　　 不过在最近一次维护网络时&#xff0c;管理员误操作使得某两台电脑之间增…
阅读更多...
vue实现相机拍摄，可录视频、拍照片、前置后置切换（简单小demo）

vue实现相机拍摄，可录视频、拍照片、前置后置切换（简单小demo）

内容比较简单&#xff0c;不做过多赘述&#xff0c;只做分享&#xff0c;测试demo&#xff0c;功能有些缺陷&#xff0c;希望路过的大佬多多指正 /(*/ω&#xff3c;*) <script setup> import { showToast, showSuccessToast, showFailToast, showLoadingToast } from …
阅读更多...
VUE3——reactive对比ref

VUE3——reactive对比ref

从定义数据角度对比: 。ref用来定义:基本类型数据 。reactive用来定义:对象(或数组)类型数据。 。备注:ref也可以用来定义对象(或数组)类型数据,它内部会自动通过 reactive 转为代理对象。 从原理角度对比: 。ref通过 object.defineProperty()的 get 与set 来实现响应式(数据劫…
阅读更多...
算法题1两数之和

算法题1两数之和

问题&#xff1a;给定一个整数数组 nums 和一个整数目标值 target&#xff0c;请你在该数组中找出 和为目标值 target 的那 两个 整数&#xff0c;并返回它们的数组下标。 你可以假设每种输入只会对应一个答案。但是&#xff0c;数组中同一个元素在答案里不能重复出现。 你可以…
阅读更多...
金三银四面试题（八）：JVM常见面试题(2)

金三银四面试题（八）：JVM常见面试题(2)

今天我们继续探讨常见的JVM面试题。这些问题不比之前的问题庞大&#xff0c;多用于面试中​JVM部分的热身运动&#xff0c;开胃菜&#xff0c;但是大家已经要认真准备。 JRE、JDK、JVM 及JIT 之间有什么不同&#xff1f; JRE 代表Java 运行时&#xff08;Java run-time&#…
阅读更多...
专升本-数字媒体

专升本-数字媒体

数字媒体 概念&#xff1a; 媒体&#xff1a;是信息的载体&#xff0c;传播信息的媒介&#xff0c;能为信息的传播提供平台 数字媒体&#xff1a;多重媒体&#xff0c;使用文字&#xff0c;数据&#xff0c;图像&#xff0c;声音等各种媒体 数字媒体技术&#xff1a;利用计…
阅读更多...
Kafka入门到实战-第四弹

Kafka入门到实战-第四弹

Kafka入门到实战 Kafka集群搭建官网地址Kafka概述使用Kraft搭建Kafka集群更新计划 Kafka集群搭建 官网地址 声明: 由于操作系统, 版本更新等原因, 文章所列内容不一定100%复现, 还要以官方信息为准 https://kafka.apache.org/Kafka概述 Apache Kafka 是一个开源的分布式事件…
阅读更多...
计算机视觉之三维重建（5）---双目立体视觉

计算机视觉之三维重建（5）---双目立体视觉

文章目录 一、平行视图1.1 示意图1.2 平行视图的基础矩阵1.3 平行视图的极几何1.4 平行视图的三角测量 二、图像校正三、对应点问题3.1 相关匹配法3.2 归一化相关匹配法3.3 窗口问题3.4 相关法存在的问题3.5 约束问题 一、平行视图 1.1 示意图 如下图即是一个平行视图。特点&a…
阅读更多...
最新文章

Copyright @ 2022~2023