前言
Mysql中ResultSet默认会将一次查询的结果存入内存中。如果数据量比较大,就会占用大量的内存。如果内存不够,就会报错。
Mysql版本
mysql-connector-java分为5.1版本和8.0版本,但有的朋友可能在项目中用到过6.0版本的jar包,而我们在官网的截图中确实没看到6.0版本。对此我也有很大的疑问,通过查看官方文档,我了解到8.0以前就是6.0,后来版本号更改,6.0变成了8.0,官方推荐升级使用8.0版本的mysql-connector-java
。
对比
5.1版本支持java5及其以上的版本,支持5.6、5.7、8.0版本的mysql数据库,支持3.0、4.0、4.1、4.2版本的jdbc。在5.1中,Driver的实现类的全路径名是
com.mysql.jdbc.Driver
。8.0版本支持java8及其以上的版本,支持5.6、5.7、8.0版本的mysql数据库,支持4.2版本的jdbc。在8.0中,Driver的实现类的全路径名是
com.mysql.cj.jdbc.Driver
。
依赖
1 | <!--操作Mysql--> |
方式1
流式处理结果,让驱动每次返回1行数据
1 | conn = DriverManager.getConnection("jdbc:mysql://192.168.7.102/", "user", "123456"); |
statement进行java.sql.ResultSet.TYPE_FORWARD_ONLY,java.sql.ResultSet.CONCUR_READ_ONLY
和stmt.setFetchSize(Integer.MIN_VALUE)
的组合设置,会告诉mysql服务器流式处理返回结果,一行一行的返回数据。
这是mysql规定的设置,一开始还搞不懂为啥setFetchSize会是Integer.MIN_VALUE,设置上就完了,这是规定的设置!
注意:当你使用此方式处理数据时,你必须处理完resultset中的所有数据,或者将resultset关闭后才能使用此连接进行下一次的查询等操作,否则会抛出异常。
方式2
使用基于游标的处理,setFetchSize
1 | import java.sql.*; |
准备数据
1 | create database ztest; |
使用HiveJDBC实现
依赖
1 | <!--JSON工具--> |
Mysql2Hive
1 | import com.alibaba.fastjson2.JSONObject; |
MysqlReader
1 | package com.xhkjedu.mysql2hive; |
ResultSet.next其实是取一条就跟数据库通讯拿一条数据,并不是全部取出放在内存,因为ResultSet.next之前,是获取了数据库连接的,数据库连接断开,你就获取不到数据了,说明是有通讯的。
HiveWriter
1 | package com.xhkjedu.mysql2hive; |
方式2(推荐)
Mysql2Hive
1 | package com.xhkjedu.mysql2hive; |
MysqlReader
1 | package com.xhkjedu.mysql2hive; |
报错
An input of GenericTypeInfo
cannot be converted to Table.
设置类型
1 | DataStream<Row> mysqlData = env.addSource(new MysqlReader(),getRowTypeInfo()); |
支持的数据类型
前面示例中的 DataStream,流中的数据类型都是定义好的 POJO 类。如果 DataStream 中的类型是简单的基本类型,还可以直接转换成表吗?这就涉及了Table 中支持的数据类型。
整体来看,DataStream 中支持的数据类型,Table 中也是都支持的,只不过在进行转换时需要注意一些细节。
1. 原子类型
在 Flink 中,基础数据类型(Integer、Double、String)和通用数据类型(也就是不可再拆分的数据类型)统一称作”原子类型”。原子类型的DataStream,转换之后就成了只有一列的Table,列字段(field)的数据类型可以由原子类型推断出。另外,还可以在 fromDataStream()方法里增加参数,用来重新命名列字段。
1 | // 将数据流转换成动态表,动态表只有一个字段,重命名为 myLong |
2. Tuple 类型
当原子类型不做重命名时,默认的字段名就是”f0”,容易想到,这其实就是将原子类型看作了一元组Tuple1 的处理结果。
Table 支持 Flink 中定义的元组类型Tuple,对应在表中字段名默认就是元组中元素的属性名 f0、f1、f2…。所有字段都可以被重新排序,也可以提取其中的一部分字段。字段还可以通过调用表达式的 as()方法来进行重命名。
1 | // 将数据流转换成只包含 f1 字段的表 |
3. POJO 类型
Flink 也支持多种数据类型组合成的”复合类型”,最典型的就是简单 Java 对象(POJO 类型)。由于 POJO 中已经定义好了可读性强的字段名,这种类型的数据流转换成 Table 就显得无比顺畅了。
将 POJO 类型的DataStream 转换成 Table,如果不指定字段名称,就会直接使用原始 POJO类型中的字段名称。POJO 中的字段同样可以被重新排序、提却和重命名,这在之前的例子中已经有过体现。
1 | Table table = tableEnv.fromDataStream(stream); |
4. Row 类型
Flink 中还定义了一个在关系型表中更加通用的数据类型——行(Row),它是 Table 中数据的基本组织形式。Row 类型也是一种复合类型,它的长度固定,而且无法直接推断出每个字段的类型,所以在使用时必须指明具体的类型信息;我们在创建 Table 时调用的 CREATE 语句就会将所有的字段名称和类型指定,这在 Flink 中被称为表的”模式结构”(Schema)。除此之外,Row 类型还附加了一个属性 RowKind,用来表示当前行在更新操作中的类型。这样, Row 就可以用来表示更新日志流(changelog stream)中的数据,从而架起了 Flink 中流和表的转换桥梁。
所以在更新日志流中,元素的类型必须是 Row,而且需要调用 ofKind()方法来指定更新类型。下面是一个具体的例子:
1 | DataStream<Row> dataStream = env.fromElements( |