1.Flink深入浅出:JDBC Connector源码分析
2.MySQL JDBC 编译添加 Maven 依赖支持
Flink深入浅出:JDBC Connector源码分析
大数据开发中,源码数据分析与报表制作是源码日常工作中最常遇到的任务。通常,源码我们通过读取Hive数据来进行计算,源码并将结果保存到数据库中,源码然后通过前端读取数据库来进行报表展示。源码autocore源码然而,源码使用FlinkSQL可以简化这一过程,源码通过一个SQL语句即可完成整个ETL流程。源码
在Flink中,源码读取Hive数据并将数据写入数据库是源码常见的需求。本文将重点讲解数据如何写入数据库的源码过程,包括刷写数据库的源码机制和原理。
以下是源码本文将讲解的几个部分,以解答在使用过程中可能产生的源码疑问:
1. 表的定义
2. 定义的表如何找到具体的实现类(如何自定义第三方sink)
3. 写入数据的机制原理
(本篇基于1..0源码整理而成)
1. 表的定义
Flink官网提供了SQL中定义表的示例,以下以oracle为例:
定义好这样的表后,就可以使用insert into student执行插入操作了。接下来,我们将探讨其中的技术细节。
2. 如何找到实现类
实际上,这一过程涉及到之前分享过的love源码SPI(服务提供者接口),即DriverManager去寻找Driver的过程。在Flink SQL执行时,会通过translate方法将SQL语句转换为对应的Operation,例如insert into xxx中的xxx会转换为CatalogSinkModifyOperation。这个操作会获取表的信息,从而得到Table对象。如果这个Table对象是CatalogTable,则会进入TableFactoryService.find()方法找到对应的实现类。
寻找实现类的过程就是SPI的过程。即通过查找路径下所有TableFactory.class的ant源码实现类,加载到内存中。这个SPI的定义位于resources下面的META-INFO下,定义接口以及实现类。
加载到内存后,首先判断是否是TableFactory的实现类,然后检查必要的参数是否满足(如果不满足会抛出异常,很多人在第一次使用Flink SQL注册表时,都会遇到NoMatchingTableFactoryException异常,其实都是因为配置的属性不全或者Jar报不满足找不到对应的TableFactory实现类造成的)。
找到对应的dwz源码实现类后,调用对应的createTableSink方法就能创建具体的实现类了。
3. 工厂模式+创建者模式,创建TableSink
JDBCTableSourceSinkFactory是JDBC表的具体实现工厂,它实现了stream的sinkfactory。在1..0版本中,它不能在batch模式下使用,但在1.版本中据说会支持。这个类使用了经典的工厂模式,其中createStreamTableSink负责创建真正的Table,基于创建者模式构建JDBCUpsertTableSink。花椒源码
创建出TableSink之后,就可以使用Flink API,基于DataStream创建一个Sink,并配置对应的并行度。
4. 消费数据写入数据库
在消费数据的过程中,底层基于PreparedStatement进行批量提交。需要注意的是提交的时机和机制。
控制刷写触发的最大数量 'connector.write.flush.max-rows' = ''
控制定时刷写的时间 'connector.write.flush.interval' = '2s'
这两个条件先到先触发,这两个参数都是可以通过with()属性配置的。
JDBCUpsertFunction很简单,主要的工作是包装对应的Format,执行它的open和invoke方法。其中open负责开启连接,invoke方法负责消费每条数据提交。
接下来,我们来看看关键的format.open()方法:
接下来就是消费数据,执行提交了
AppendWriter很简单,只是对PreparedStatement的封装而已
5. 总结
通过研究代码,我们应该了解了以下关键问题:
1. JDBC Sink执行的机制,比如依赖哪些包?(flink-jdbc.jar,这个包提供了JDBCTableSinkFactory的实现)
2. 如何找到对应的实现?基于SPI服务发现,扫描接口实现类,通过属性过滤,最终确定对应的实现类。
3. 底层如何提交记录?目前只支持append模式,底层基于PreparedStatement的addbatch+executeBatch批量提交
4. 数据写入数据库的时机和机制?一方面定时任务定时刷新,另一方面数量超过限制也会触发刷新。
更多Flink内容参考:
MySQL JDBC 编译添加 Maven 依赖支持
在当前的工作项目中,需要对MySQL JDBC进行编译,即集成mysql-connector-j包。进入年,我们依然面临着手动下载和安装JAR包的传统方式,这从MySQL官方文档的JDBC源码编译指南中可见一斑。Oracle的这一做法似乎有意为之,给MySQL开发者带来了不便。
为了解决这个问题,我决定将MySQL JDBC添加Maven依赖,以下是关键步骤:
首先,确保你的项目配置了JUnit 5进行单元测试,这时需要在pom.xml中加入maven-surefire-plugin插件:
xml
org.apache.maven.plugins
maven-surefire-plugin
如果你希望尽快使用,而不是等待官方更新,可以直接从我fork的库中获取,选择feat-maven-dep分支。
虽然这个过程略显繁琐,但通过这种方式,我们至少可以简化构建流程,提高开发效率。期待MySQL官方能尽快采纳这些改进。