【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依赖

Table/SQL Api 扩展依赖
Table/SQL Api 基础依赖
Table/SQL Api 和 DataStream Api 交互的依赖 bridge
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）

Flink读写MySQL工具类

Table Api 环境加载

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

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

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

Planner（查询处理器）

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

Flink Planner的类型：

Flink Planner (Old Planner)
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版本之前）

PG (PostgresSQL) Catalog
HiveCatalog
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连接的url
        String 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表

创建与表映射的实体类

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

class MysqlUtilTest {

    @DisplayName("测试使用 Flink Table/SQL Api 读取Mysql")
    @Test
    public void testReadWithTableOrSQLApi() throws Exception {
        // 初始化环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment();
        // 设置并行度1
        env.setParallelism(1);
        // 获取参数工具实例
        ParameterTool parameterTool = ParameterUtil.getParameters();

        /* **********************
         *
         * CREATE 语句用于向当前或指定的 Catalog 中注册表。
         * 注册后的表、视图和函数可以在 SQL 查询中使用
         *
         * *********************/
        // 表名
        String tableName = "tb_user";

        // 表字段ddl
        String 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();
    }
}