怎么使用Tbale SQL与Flink JDBC连接器将数据插入MYSQL数据库表

这篇文章主要介绍“怎么使用Tbale SQL与Flink JDBC连接器将数据插入MYSQL数据库表”，在日常操作中，相信很多人在怎么使用Tbale SQL与Flink JDBC连接器将数据插入MYSQL数据库表问题上存在疑惑，小编查阅了各式资料，整理出简单好用的操作方法，希望对大家解答”怎么使用Tbale SQL与Flink JDBC连接器将数据插入MYSQL数据库表”的疑惑有所帮助！接下来，请跟着小编一起来学习吧！

示例环境

java.version: 1.8.xflink.version: 1.11.1kafka：2.11

InsertToMysql.java

package com.flink.examples.mysql; import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.StatementSet; import org.apache.flink.table.api.TableResult; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; /**  * @Description 使用Tbale&SQL与Flink JDBC连接器将数据插入MYSQL数据库表  */ public class InsertToMysql {     /**      官方参考：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html      */     //flink-jdbc-1.11.1写法,所有属性名在JdbcTableSourceSinkFactory工厂类中定义     static String table_sql =             "CREATE TABLE my_users (\n" +             "  id BIGINT,\n" +             "  name STRING,\n" +             "  age INT,\n" +             "  status INT,\n" +             "  PRIMARY KEY (id) NOT ENFORCED\n" +             ") WITH (\n" +             "  'connector.type' = 'jdbc',\n" +             "  'connector.url' = 'jdbc:mysql://192.168.110.35:3306/flink?useUnicode=true&characterEncoding=utf-8', -- 数据库需建立 \n" +             "  'connector.driver' = 'com.mysql.jdbc.Driver', \n" +             "  'connector.table' = 'users', -- 已知的表 \n" +             "  'connector.username' = 'root',\n" +             "  'connector.password' = 'password' \n" +             ")";     public static void main(String[] args) throws Exception {         //构建StreamExecutionEnvironment         StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();         //构建EnvironmentSettings 并指定Blink Planner         EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();         //构建StreamTableEnvironment         StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);         //注册mysql数据维表         tEnv.executeSql(table_sql);         //执行SQL,id=0是因id字段为自增主键，为0则mysql识别会默认自增值代替         String sql = "insert into my_users (id,name,age,status) values(0,'tom',31,0)";         // 第一种方式：直接执行sql //        TableResult tableResult = tEnv.executeSql(sql);         //第二种方式：声明一个操作集合来执行sql         StatementSet stmtSet = tEnv.createStatementSet();         stmtSet.addInsertSql(sql);         TableResult tableResult = stmtSet.execute();         tableResult.print();     } }

 打印结果

+-------------------------------------------+ | default_catalog.default_database.my_users | +-------------------------------------------+ |                                        -1 | +-------------------------------------------+ 1 row in set

到此，关于“怎么使用Tbale SQL与Flink JDBC连接器将数据插入MYSQL数据库表”的学习就结束了，希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习，快去试试吧！若想继续学习更多相关知识，请继续关注亿速云网站，小编会继续努力为大家带来更多实用的文章！