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

1. 导言：

Apache Flink是一款功能强大的流式处理引擎，可用于实时处理大规模数据。本文将先容怎样使用Flink与MySQL数据库进行交互，以洗濯股票数据为例。
2. 情况准备：

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

1. CREATE TABLE `re_stock_code_price` (
2.   `id` bigint NOT NULL AUTO_INCREMENT,
3.   `code` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票代码',
4.   `name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票名称',
5.   `close` double DEFAULT NULL COMMENT '最新价',
6.   `change_percent` double DEFAULT NULL COMMENT '涨跌幅',
7.   `change` double DEFAULT NULL COMMENT '涨跌额',
8.   `volume` double DEFAULT NULL COMMENT '成交量（手）',
9.   `amount` double DEFAULT NULL COMMENT '成交额',
10.   `amplitude` double DEFAULT NULL COMMENT '振幅',
11.   `turnover_rate` double DEFAULT NULL COMMENT '换手率',
12.   `peration` double DEFAULT NULL COMMENT '市盈率',
13.   `volume_rate` double DEFAULT NULL COMMENT '量比',
14.   `hign` double DEFAULT NULL COMMENT '最高',
15.   `low` double DEFAULT NULL COMMENT '最低',
16.   `open` double DEFAULT NULL COMMENT '今开',
17.   `previous_close` double DEFAULT NULL COMMENT '昨收',
18.   `pb` double DEFAULT NULL COMMENT '市净率',
19.   `create_time` varchar(64) NOT NULL COMMENT '写入时间',
20.   `rise` int NOT NULL,
21.   PRIMARY KEY (`id`)
22. ) ENGINE=InnoDB AUTO_INCREMENT=11207 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
24. CREATE TABLE `t_stock_code_price` (
25.   `id` bigint NOT NULL AUTO_INCREMENT,
26.   `code` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票代码',
27.   `name` varchar(64) CHARACTER SET utf8mb4 COLLATE utf8mb4_0900_ai_ci NOT NULL COMMENT '股票名称',
28.   `close` double DEFAULT NULL COMMENT '最新价',
29.   `change_percent` double DEFAULT NULL COMMENT '涨跌幅',
30.   `change` double DEFAULT NULL COMMENT '涨跌额',
31.   `volume` double DEFAULT NULL COMMENT '成交量（手）',
32.   `amount` double DEFAULT NULL COMMENT '成交额',
33.   `amplitude` double DEFAULT NULL COMMENT '振幅',
34.   `turnover_rate` double DEFAULT NULL COMMENT '换手率',
35.   `peration` double DEFAULT NULL COMMENT '市盈率',
36.   `volume_rate` double DEFAULT NULL COMMENT '量比',
37.   `hign` double DEFAULT NULL COMMENT '最高',
38.   `low` double DEFAULT NULL COMMENT '最低',
39.   `open` double DEFAULT NULL COMMENT '今开',
40.   `previous_close` double DEFAULT NULL COMMENT '昨收',
41.   `pb` double DEFAULT NULL COMMENT '市净率',
42.   `create_time` varchar(64) NOT NULL COMMENT '写入时间',
43.   PRIMARY KEY (`id`)
44. ) ENGINE=InnoDB AUTO_INCREMENT=11207 DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci

复制代码

1. package org.east;
3. import org.apache.flink.api.common.RuntimeExecutionMode;
4. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment;
5. import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment;
7. object TableETL {
8.     def main(args: Array[String]): Unit = {
9.         val senv = StreamExecutionEnvironment.getExecutionEnvironment
10.             .setRuntimeMode(RuntimeExecutionMode.STREAMING)
11.         val tEnv = StreamTableEnvironment.create(senv)
13.         // 定义源表
14.         val source_table =
15.             """
16.             CREATE TEMPORARY TABLE t_stock_code_price (
17.                 id BIGINT NOT NULL,
18.                 code STRING NOT NULL,
19.                 -- 其他字段...
20.                 create_time STRING NOT NULL,
21.                 PRIMARY KEY (id) NOT ENFORCED
22.             ) WITH (
23.                 'connector' = 'jdbc',
24.                 'url' = 'jdbc:mysql://localhost:3306/mydb',
25.                 'driver' = 'com.mysql.cj.jdbc.Driver',
26.                 'table-name' = 't_stock_code_price',
27.                 'username' = 'root',
28.                 'password' = '12345678'
29.             )
30.             """.stripMargin
32.         // 定义目标表
33.         val sink_table =
34.             """
35.             CREATE TEMPORARY TABLE re_stock_code_price (
36.                 id BIGINT NOT NULL,
37.                 code STRING NOT NULL,
38.                 -- 其他字段...
39.                 create_time STRING NOT NULL,
40.                 rise INT,
41.                 PRIMARY KEY (id) NOT ENFORCED
42.             ) WITH (
43.                 'connector' = 'jdbc',
44.                 'url' = 'jdbc:mysql://localhost:3306/mydb',
45.                 'driver' = 'com.mysql.cj.jdbc.Driver',
46.                 'table-name' = 're_stock_code_price',
47.                 'username' = 'root',
48.                 'password' = '12345678'
49.             )
50.             """.stripMargin
52.         tEnv.executeSql(source_table)
53.         tEnv.executeSql(sink_table)

复制代码

在这段代码中，我们起首创建了Flink的流式执行情况和StreamTableEnvironment。然后，我们定义了两个临时表，用于存储原始股票数据和洗濯后的数据。
3. 数据洗濯：

接下来，我们执行数据洗濯操纵，并将结果写入目的表。

1.         // 执行清洗操作，并将结果写入目标表
2.         tEnv.executeSql("INSERT INTO re_stock_code_price " +
3.             "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. 完备代码：

下面是完备的代码：

1. package org.east;
2. import org.apache.flink.api.common.RuntimeExecutionMode
3. import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
4. import org.apache.flink.table.api.bridge.scala.StreamTableEnvironment
6. object TableETL {
8.   def main(args: Array[String]): Unit = {
10.     val senv = StreamExecutionEnvironment.getExecutionEnvironment
11.       .setRuntimeMode(RuntimeExecutionMode.STREAMING)
12.     val tEnv = StreamTableEnvironment.create(senv)
14.     val source_table =
15.       """
16.         |CREATE TEMPORARY TABLE t_stock_code_price (
17.         |  id BIGINT NOT NULL,
18.         |  code STRING NOT NULL,
19.         |  name STRING NOT NULL,
20.         |  `close` DOUBLE,
21.         |  change_percent DOUBLE,
22.         |  change DOUBLE,
23.         |  volume DOUBLE,
24.         |  amount DOUBLE,
25.         |  amplitude DOUBLE,
26.         |  turnover_rate DOUBLE,
27.         |  peration DOUBLE,
28.         |  volume_rate DOUBLE,
29.         |  hign DOUBLE,
30.         |  low DOUBLE,
31.         |  `open` DOUBLE,
32.         |  previous_close DOUBLE,
33.         |  pb DOUBLE,
34.         |  create_time STRING NOT NULL,
35.         |  PRIMARY KEY (id) NOT ENFORCED
36.         |) WITH (
37.         |   'connector' = 'jdbc',
38.         |   'url' = 'jdbc:mysql://localhost:3306/mydb',
39.         |   'driver' = 'com.mysql.cj.jdbc.Driver',
40.         |   'table-name' = 't_stock_code_price',
41.         |   'username' = 'root',
42.         |   'password' = '12345678'
43.         |)
44.         |""".stripMargin
47.     val sink_table =
48.       """
49.         |CREATE TEMPORARY TABLE re_stock_code_price (
50.         |  id BIGINT NOT NULL,
51.         |  code STRING NOT NULL,
52.         |  name STRING NOT NULL,
53.         |  `close` DOUBLE,
54.         |  change_percent DOUBLE,
55.         |  change DOUBLE,
56.         |  volume DOUBLE,
57.         |  amount DOUBLE,
58.         |  amplitude DOUBLE,
59.         |  turnover_rate DOUBLE,
60.         |  peration DOUBLE,
61.         |  volume_rate DOUBLE,
62.         |  hign DOUBLE,
63.         |  low DOUBLE,
64.         |  `open` DOUBLE,
65.         |  previous_close DOUBLE,
66.         |  pb DOUBLE,
67.         |  create_time STRING NOT NULL,
68.         |  rise int,
69.         |  PRIMARY KEY (id) NOT ENFORCED
70.         |) WITH (
71.         |   'connector' = 'jdbc',
72.         |   'url' = 'jdbc:mysql://localhost:3306/mydb',
73.         |   'driver' = 'com.mysql.cj.jdbc.Driver',
74.         |   'table-name' = 're_stock_code_price',
75.         |   'username' = 'root',
76.         |   'password' = '12345678'
77.         |)
78.         |""".stripMargin
81.     tEnv.executeSql(source_table)
82.     tEnv.executeSql(sink_table)
83.     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")
85.     val user_DS = tEnv.executeSql("select * from re_stock_code_price")
87.     user_DS.print()
88.   }
89. }

复制代码

如有遇到问题可以找小编沟通交换哦。别的小编帮忙辅导大课作业，学生毕设等。不限于python，java，大数据，模型练习等。

免责声明：如果侵犯了您的权益，请联系站长，我们会及时删除侵权内容，谢谢合作！更多信息从访问主页：qidao123.com:ToB企服之家，中国第一个企服评测及商务社交产业平台。