【Table/SQL Api】Flink Table/SQL Api表转流读取MySQL

2023-12-10 22:28
文章标签 sql mysql 读取 api table flink database 转流

本文主要是介绍【Table/SQL Api】Flink Table/SQL Api表转流读取MySQL，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

引入依赖

jdbc依赖

flink-connector-jdbc + mysql-jdbc-driver 操作mysql数据库

        <!-- Flink-Connector-Jdbc --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_${scala.binary.version}</artifactId></dependency><!-- mysql jdbc driver --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId></dependency>

Table/SQL Api依赖

  1. Table/SQL Api 扩展依赖
  2. Table/SQL Api 基础依赖
  3. Table/SQL Api 和 DataStream Api 交互的依赖 bridge
  4. Flink Planner 依赖
        <!-- Table/SQL Api 依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java</artifactId></dependency><!-- Table/SQL Api 扩展依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-common</artifactId></dependency><!-- bridge桥接器，主要负责Table API和 DataStream API的连接支持 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId></dependency><!-- Flink Planner 依赖 --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_${scala.binary.version}</artifactId></dependency>

对应版本在这 （项目Flink版本为1.14.5）

image-20231210161727111

Flink读写MySQL工具类

Table Api 环境加载

Table API和SQL Api都是基于Table接口

Table Api上下文环境有3种类型：

  1. TableEnvironment：只支持Batch作业
  2. BatchTableEnvironment：只支持Batch作业
  3. StreamTableEnvironment: 支持流计算【用这个】

Planner（查询处理器）

Planner（查询处理器）：解析sql、优化sql和执行sql

Flink Planner的类型：

  1. Flink Planner (Old Planner)
  2. Blink Planner （Flink 1.14之前需要手动导入依赖）

Blink Planner从Flink 1.11版本开始为Flink-table的默认查询处理器

Blink Planner使得Table Api & SQL 层实现了流批统一

Catalog对象

Catalog对象是提供了元数据信息，数据源与数据表的信息则存储在Catalog中

// 创建Catalog对象
new JdbcCatalog(catalog_name, database, username, passwd, url);

Catalog对象是接口

Catalog接口的实现：（Flink 1.14版本之前）

  1. PG (PostgresSQL) Catalog
  2. HiveCatalog
  3. Mysql Catalog (Flink 1.15 才有)

DDL与数据库表结构必须一模一样，建立映射，这种方式数据库表结构如果变化，代码也必须随之变化重新打包，因此这种方式用的不多，一般catalog会用的比较多。

但由于项目Flink依赖用的是1.14.5，因此还是使用DDL语句实现。

代码实现

public class MysqlUtil {/*** 数据库连接对象*/private static Connection connection = null;/*** SQL语句对象*/private static PreparedStatement preparedStatement = null;/*** 结果集对象*/private static ResultSet rs = null;/*** 使用 Flink Table/SQL Api 读取Mysql** @param env:           流计算上下文环境* @param parameterTool: 参数工具* @param clazz:         流水线输出对象的类* @param tableName:     表名* @param ddlString:     DDL字符串* @param sql:           SQL查询语句* @return DataStream<T>：DataStream对象*/public static <T> DataStream<T> readWithTableOrSQLApi(StreamExecutionEnvironment env,ParameterTool parameterTool,Class<T> clazz,String tableName,String ddlString,String sql) throws Exception {// 创建TableApi运行环境EnvironmentSettings bsSettings =EnvironmentSettings.newInstance()// Flink 1.14不需要再设置 Planner//.useBlinkPlanner()// 设置流计算模式.inStreamingMode().build();// 创建StreamTableEnvironment实例StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, bsSettings);// 指定方言 (选择使用SQL语法还是HQL语法)tableEnv.getConfig().setSqlDialect(SqlDialect.DEFAULT);// 编写DDL ( 数据定义语言 )String ddl = buildMysqlDDL(parameterTool, tableName, ddlString);// StreamTableEnvironment注册虚拟表tableEnv.executeSql(ddl);// 查询结果是Table对象Table table = tableEnv.sqlQuery(sql);// 将Table对象转换为DataStream对象return tableEnv.toDataStream(table, clazz);}/*** 根据参数生成MySQL的DDL语句** @param parameterTool  参数工具，用于获取MySQL连接信息* @param tableName      要创建的表名* @param ddlFieldString 表字段的DDL语句* @return 生成的完整的MySQL DDL语句*/public static String buildMysqlDDL(ParameterTool parameterTool,String tableName,String ddlFieldString) {// 从参数工具中获取mysql连接的urlString url = parameterTool.get(ParameterConstants.Mysql_URL);// 从参数工具中获取mysql连接的用户名String username = parameterTool.get(ParameterConstants.Mysql_USERNAME);// 从参数工具中获取mysql连接的密码String passwd = parameterTool.get(ParameterConstants.Mysql_PASSWD);// 从参数工具中获取MySQL的驱动程序String driver = parameterTool.get(ParameterConstants.Mysql_DRIVER);// 返回完整的DDL语句return "CREATE TABLE IF NOT EXISTS " +tableName +" (\n" +ddlFieldString +")" +" WITH (\n" +"'connector' = 'jdbc',\n" +"'driver' = '" + driver + "',\n" +"'url' = '" + url + "',\n" +"'username' = '" + username + "',\n" +"'password' = '" + passwd + "',\n" +"'table-name' = '" + tableName + "'\n" +")";}/*** 初始化 jdbc Connection*/public static Connection init(ParameterTool parameterTool) {String _url = parameterTool.get(ParameterConstants.Mysql_URL);String _username = parameterTool.get(ParameterConstants.Mysql_USERNAME);String _passwd = parameterTool.get(ParameterConstants.Mysql_PASSWD);try {connection = DriverManager.getConnection(_url, _username, _passwd);} catch (Exception e) {throw new RuntimeException(e);}return connection;}/*** 生成 PreparedStatement*/public static PreparedStatement initPreparedStatement(String sql) {try {preparedStatement = connection.prepareStatement(sql);} catch (Exception e) {throw new RuntimeException(e);}return preparedStatement;}/*** 关闭 jdbc Connection*/public static void close() {try {if (preparedStatement != null) {preparedStatement.close();}if (connection != null) {connection.close();}} catch (Exception e) {throw new RuntimeException(e);}}/*** 关闭 PreparedStatement*/public static void closePreparedStatement() {try {if (preparedStatement != null) {preparedStatement.close();}} catch (Exception e) {throw new RuntimeException(e);}}/*** 关闭 ResultSet*/public static void closeResultSet() {try {if (rs != null) {rs.close();}} catch (Exception e) {throw new RuntimeException(e);}}/*** 执行 sql 语句*/public static ResultSet executeQuery(PreparedStatement ps) {preparedStatement = ps;try {rs = preparedStatement.executeQuery();} catch (Exception e) {throw new RuntimeException(e);}return rs;}}

测试一下

测试库中有个tb_user表

image-20231210174346826

创建与表映射的实体类

@Data
public class UserPO {private Long id;private String name;
}

class MysqlUtilTest {@DisplayName("测试使用 Flink Table/SQL Api 读取Mysql")@Testpublic void testReadWithTableOrSQLApi() throws Exception {// 初始化环境StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();// 设置并行度1env.setParallelism(1);// 获取参数工具实例ParameterTool parameterTool = ParameterUtil.getParameters();/* ************************ CREATE 语句用于向当前或指定的 Catalog 中注册表。* 注册后的表、视图和函数可以在 SQL 查询中使用** *********************/// 表名String tableName = "tb_user";// 表字段ddlString ddlFieldString ="id BIGINT,\n" +"name STRING \n";// 查询表的全部字段String sql = "SELECT * FROM " + tableName;DataStream<UserPO> rowDataStream =MysqlUtil.readWithTableOrSQLApi(env,parameterTool,UserPO.class,tableName,ddlFieldString,sql);rowDataStream.print("mysql");env.execute();}
}

image-20231210174720832

查询成功！

这篇关于【Table/SQL Api】Flink Table/SQL Api表转流读取MySQL的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/478575。

相关文章

MySQL的JDBC编程详解

MySQL的JDBC编程详解

《MySQL的JDBC编程详解》：本文主要介绍MySQL的JDBC编程,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录前言一、前置知识1. 引入依赖2. 认识 url二、JDBC 操作流程1. JDBC 的写操作2. JDBC 的读操作总结前言本文介绍了mysq
阅读更多...
java.sql.SQLTransientConnectionException连接超时异常原因及解决方案

java.sql.SQLTransientConnectionException连接超时异常原因及解决方案

《java.sql.SQLTransientConnectionException连接超时异常原因及解决方案》：本文主要介绍java.sql.SQLTransientConnectionExcep... 目录一、引言二、异常信息分析三、可能的原因3.1 连接池配置不合理3.2 数据库负载过高3.3 连接泄漏
阅读更多...
Linux下MySQL数据库定时备份脚本与Crontab配置教学

Linux下MySQL数据库定时备份脚本与Crontab配置教学

《Linux下MySQL数据库定时备份脚本与Crontab配置教学》在生产环境中,数据库是核心资产之一,定期备份数据库可以有效防止意外数据丢失,本文将分享一份MySQL定时备份脚本,并讲解如何通过cr... 目录备份脚本详解脚本功能说明授权与可执行权限使用 Crontab 定时执行编辑 Crontab添加定
阅读更多...
PHP应用中处理限流和API节流的最佳实践

PHP应用中处理限流和API节流的最佳实践

《PHP应用中处理限流和API节流的最佳实践》限流和API节流对于确保Web应用程序的可靠性、安全性和可扩展性至关重要,本文将详细介绍PHP应用中处理限流和API节流的最佳实践,下面就来和小编一起学习... 目录限流的重要性在 php 中实施限流的最佳实践使用集中式存储进行状态管理（如 Redis）采用滑动
阅读更多...
MySQL中On duplicate key update的实现示例

MySQL中On duplicate key update的实现示例

《MySQL中Onduplicatekeyupdate的实现示例》ONDUPLICATEKEYUPDATE是一种MySQL的语法,它在插入新数据时,如果遇到唯一键冲突,则会执行更新操作,而不是抛... 目录1/ ON DUPLICATE KEY UPDATE的简介2/ ON DUPLICATE KEY UP
阅读更多...
MySQL分库分表的实践示例

MySQL分库分表的实践示例

《MySQL分库分表的实践示例》MySQL分库分表适用于数据量大或并发压力高的场景,核心技术包括水平/垂直分片和分库,需应对分布式事务、跨库查询等挑战,通过中间件和解决方案实现,最佳实践为合理策略、备... 目录一、分库分表的触发条件1.1 数据量阈值1.2 并发压力二、分库分表的核心技术模块2.1 水平分
阅读更多...
Python与MySQL实现数据库实时同步的详细步骤

Python与MySQL实现数据库实时同步的详细步骤

《Python与MySQL实现数据库实时同步的详细步骤》在日常开发中,数据同步是一项常见的需求,本篇文章将使用Python和MySQL来实现数据库实时同步,我们将围绕数据变更捕获、数据处理和数据写入这... 目录前言摘要概述：数据同步方案1. 基本思路2. mysql Binlog 简介实现步骤与代码示例1
阅读更多...
使用shardingsphere实现mysql数据库分片方式

使用shardingsphere实现mysql数据库分片方式

《使用shardingsphere实现mysql数据库分片方式》本文介绍如何使用ShardingSphere-JDBC在SpringBoot中实现MySQL水平分库,涵盖分片策略、路由算法及零侵入配置... 目录一、ShardingSphere 简介1.1 对比1.2 核心概念1.3 Sharding-Sp
阅读更多...
MySQL 表空却 ibd 文件过大的问题及解决方法

MySQL 表空却 ibd 文件过大的问题及解决方法

《MySQL表空却ibd文件过大的问题及解决方法》本文给大家介绍MySQL表空却ibd文件过大的问题及解决方法,本文给大家介绍的非常详细,对大家的学习或工作具有一定的参考借鉴价值,需要的朋友参考... 目录一、问题背景：表空却 “吃满” 磁盘的怪事二、问题复现：一步步编程还原异常场景1. 准备测试源表与数据
阅读更多...
Mac电脑如何通过 IntelliJ IDEA 远程连接 MySQL

Mac电脑如何通过 IntelliJ IDEA 远程连接 MySQL

《Mac电脑如何通过IntelliJIDEA远程连接MySQL》本文详解Mac通过IntelliJIDEA远程连接MySQL的步骤,本文通过图文并茂的形式给大家介绍的非常详细,感兴趣的朋友跟... 目录MAC电脑通过 IntelliJ IDEA 远程连接 mysql 的详细教程一、前缀条件确认二、打开 ID
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2