怎么使用TbaleSQL与FlinkJDBC连接器将数据插入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数据库表”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!
新闻标题:怎么使用TbaleSQL与FlinkJDBC连接器将数据插入MYSQL数据库表
新闻来源:http://ybzwz.com/article/iiiess.html