使用Apache Flink实现MySQL数据读取和写入的完整指南

1. 导言：

Apache Flink是一款功能强大的流式处理引擎，可用于实时处理大规模数据。本文将介绍如何使用Flink与MySQL数据库进行交互，以清洗股票数据为例。

2. 环境准备：

首先，确保已安装Apache Flink并配置好MySQL数据库。导入相关依赖包，并创建必要的Table。同时需要提前创建好mysql表，一行source表，一张sink表。

CREATE TABLE `re_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 '写入时间',`rise` int NOT NULL,PRIMARY KEY (`id`)
) ENGINE=InnoDB AUTO_INCREMENT=11207 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci 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=11207 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci 

package org.east;import org.apache.flink.api.common.RuntimeExecutionMode;
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment;object TableETL {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,-- 其他字段...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')""".stripMargin// 定义目标表val sink_table ="""CREATE TEMPORARY TABLE re_stock_code_price (id BIGINT NOT NULL,code STRING NOT NULL,-- 其他字段...create_time STRING NOT NULL,rise INT,PRIMARY KEY (id) NOT ENFORCED) WITH ('connector' = 'jdbc','url' = 'jdbc:mysql://localhost:3306/mydb','driver' = 'com.mysql.cj.jdbc.Driver','table-name' = 're_stock_code_price','username' = 'root','password' = '12345678')""".stripMargintEnv.executeSql(source_table)tEnv.executeSql(sink_table)

在这段代码中，我们首先创建了Flink的流式执行环境和StreamTableEnvironment。然后，我们定义了两个临时表，用于存储原始股票数据和清洗后的数据。

3. 数据清洗：

接下来，我们执行数据清洗操作，并将结果写入目标表。

        // 执行清洗操作，并将结果写入目标表tEnv.executeSql("INSERT INTO re_stock_code_price " +"SELECT *, CASE WHEN change_percent > 0 THEN 1 ELSE 0 END AS rise FROM t_stock_code_price")

在这里，我们计算了股票涨跌情况，并将结果写入到目标表中。在这个例子中，我们假设change_percent字段表示股票价格的变化百分比，rise字段为1表示股票上涨，为0表示股票下跌。

4. 结果展示：

最后，我们查询目标表并打印结果。

在这里插入图片描述
在这里插入图片描述

5. 完整代码：

下面是完整的代码：

package org.east;
import org.apache.flink.api.common.RuntimeExecutionMode
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.table.api.bridge.scala.StreamTableEnvironmentobject TableETL {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'|)|""".stripMarginval sink_table ="""|CREATE TEMPORARY TABLE re_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,|  rise int,|  PRIMARY KEY (id) NOT ENFORCED|) WITH (|   'connector' = 'jdbc',|   'url' = 'jdbc:mysql://localhost:3306/mydb',|   'driver' = 'com.mysql.cj.jdbc.Driver',|   'table-name' = 're_stock_code_price',|   'username' = 'root',|   'password' = '12345678'|)|""".stripMargintEnv.executeSql(source_table)tEnv.executeSql(sink_table)tEnv.executeSql("insert into re_stock_code_price select *,case when change_percent>0 then 1 else 0 end as rise from t_stock_code_price")val user_DS = tEnv.executeSql("select * from re_stock_code_price")user_DS.print()}
}

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

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

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

相关文章

如何为企业策划一场XR虚拟直播？

如何为企业策划一场XR虚拟直播？

活动年年办，都是老一套，想玩点新花样？ 预算有限，但还是想把活动办的逼格高一点？ 想通过活动，让更多的人知道自己企业的品牌？ 随着AIGC技术的不断演变，企业活动的形式和内容也在不…
阅读更多...
kprobe/kretprobe的介绍,原理图,运行情况,用户层代码+内核层代码+预编译代码(详细解释+语法介绍),修改.bpf.c入口处的函数

kprobe/kretprobe的介绍,原理图,运行情况,用户层代码+内核层代码+预编译代码(详细解释+语法介绍),修改.bpf.c入口处的函数

目录 kprobe/kretprobe 介绍 原理图 运行情况 代码解释 .bpf.c 源码 语法 / 函数接口 char LICENSE[] SEC("license") "Dual BSD/GPL" SEC do_unlinkat filename结构体 ​编辑 BPF_KPROBE bpf_get_current_pid_tgid(…
阅读更多...
Mybatis别名 动态sql语句 分页查询

Mybatis别名 动态sql语句 分页查询

给Mybatis的实体类起别名 给Mybatis的xml文件注册mapper映射文件 动态sql语句 1 if 2 choose 3 where 4 foreach 一&#xff09;if 查询指定名称商品信息 语法&#xff1a; SELECT * FROM goods where 11 <if test "gName!null"> and g.g_name like co…
阅读更多...
学习vue3第十二节（组件的使用与类型）

学习vue3第十二节（组件的使用与类型）

1、组件的作用用途 目的&#xff1a; 提高代码的复用度&#xff0c;和便于维护&#xff0c;通过封装将复杂的功能代码拆分为更小的模块&#xff0c;方便管理&#xff0c; 当我们需要实现相同的功能时&#xff0c;我们只需要复用已经封装好的组件&#xff0c;而不需要重新编写相…
阅读更多...
数据结构：基于数组实现栈

数据结构：基于数组实现栈

1 前言 栈是一种先进后出的线性表。向一个栈插入新元素可以叫做进栈、入栈、压栈&#xff0c;新元素必须放到栈顶元素上面&#xff0c;使之成为新的栈顶&#xff1b;从一个栈删除元素可以叫做出栈、退栈&#xff0c;它将栈顶元素删除&#xff0c;使和原来栈顶元素相邻的元素称…
阅读更多...
RabbitMQ 实验消费原始队列消息， 拒绝（reject）投递死信交换机过程

RabbitMQ 实验消费原始队列消息， 拒绝（reject）投递死信交换机过程

如果你想通过 RabbitMQ 的死信队列功能实现消费者拒绝消息投递到死信交换机的行为&#xff0c;你可以按照以下步骤操作&#xff1a; 创建原始队列&#xff0c;并将其绑定到一个交换机上&#xff1a; export RABBITMQ_SERVER127.0.0.1 export RABBITMQ_PORT5672 export RAB…
阅读更多...
Apache Hive的基本使用语法（二）

Apache Hive的基本使用语法（二）

Hive SQL操作 7、修改表 表重命名 alter table score4 rename to score5;修改表属性值 # 修改内外表属性 ALTER TABLE table_name SET TBLPROPERTIES("EXTERNAL""TRUE"); # 修改表注释 ALTER TABLE table_name SET TBLPROPERTIES (comment new_commen…
阅读更多...
Tomcat配置https

Tomcat配置https

前言&#xff1a;本文内容为实操记录&#xff0c;仅供参考&#xff01; 一、证书 CA证书申请下载不赘述了。 二、上传证书 进入tomcat根目录&#xff0c;conf同级目录下创建cert文件夹&#xff0c;并将证书两个文件上传到该文件夹&#xff1b; 三、编辑conf/server.xml文件 ① …
阅读更多...
20221124 kafka实时数据写入Redis

20221124 kafka实时数据写入Redis

一、上线结论 实现了将用户线上实时浏览的沉浸式视频信息&#xff0c;保存在Redis中这样一个功能。为实现沉浸式视频离线推荐到实时推荐提供了强有力的支持。目前只是应用在沉浸式场景&#xff0c;后续也能扩展到其他所有场景。用于两个场景&#xff1a;&#xff08;1&#xf…
阅读更多...
WPF自定义Panel：让拖拽变得更简单

WPF自定义Panel：让拖拽变得更简单

在 WPF 应用程序中&#xff0c;拖放操作是实现用户交互的重要组成部分。通过拖放操作&#xff0c;用户可以轻松地将数据从一个位置移动到另一个位置&#xff0c;或者将控件从一个容器移动到另一个容器。然而&#xff0c;WPF 中默认的拖放操作可能并不是那么好用。为了解决这个问…
阅读更多...
Vue中v-for多个Echarts图表组件只渲染一个要素问题排查

Vue中v-for多个Echarts图表组件只渲染一个要素问题排查

这个系列主要是用于记录我日常工作中遇到的一些Bug,既属于知识分享&#xff0c;也是对学习习惯的维持… 问题描述 今天&#xff0c;在开发一个WebGIS大屏项目时&#xff0c;我遇到了多个三维Echarts饼图图表渲染的问题&#xff0c;因为相似图表很多&#xff0c;我决定将Echart图…
阅读更多...
Unity3d使用Jenkins自动化打包（Windows）（二）

Unity3d使用Jenkins自动化打包（Windows）（二）

文章目录 前言一、Unity工程准备二、Unity调取命令行实战一实战二实战三实战四实战五 总结 前言 自动化打包的价值在于让程序员更轻松地创建和管理构建工具链&#xff0c;提高编程效率&#xff0c;将繁杂的工作碎片化&#xff0c;变成人人&#xff08;游戏行业特指策划&#x…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023