Shulou(Shulou.com)06/01 Report--

This article mainly introduces "how to insert data into MYSQL database tables using Tbale SQL and Flink JDBC connectors". In daily operation, it is believed that many people have doubts about how to use Tbale SQL and Flink JDBC connectors to insert data into MYSQL database tables. Xiaobian consulted all kinds of materials and sorted out simple and useful operation methods. I hope it will be helpful to answer the question of "how to use Tbale SQL and Flink JDBC connectors to insert data into MYSQL database tables"! Next, please follow the editor to study!

Sample environment

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 uses Tbale&SQL and Flink JDBC connectors to insert data into MYSQL database tables * / public class InsertToMysql {/ * * official reference: https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html * / / flink-jdbc-1.11.1 All attribute names are defined in the JdbcTableSourceSinkFactory factory class 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' -- Database needs to be established\ n "+" 'connector.driver' =' com.mysql.jdbc.Driver',\ n "+" 'connector.table' =' users',-- known tables\ n "+"'connector.username' =' root',\ n" + "'connector.password' =' password'\ n" + ")" Public static void main (String [] args) throws Exception {/ / build StreamExecutionEnvironment StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment (); / / build EnvironmentSettings and specify BlinkPlanner EnvironmentSettings bsSettings = EnvironmentSettings.newInstance (). UseBlinkPlanner (). InStreamingMode (). Build (); / / build StreamTableEnvironment StreamTableEnvironment tEnv = StreamTableEnvironment.create (env, bsSettings); / / register mysql data dimension table tEnv.executeSql (table_sql) / / execute SQL,id=0 because the id field is a self-increasing primary key, and if 0, mysql recognition will default to self-increment instead of String sql = "insert into my_users (id,name,age,status) values"; / / the first way: directly execute sql// TableResult tableResult = tEnv.executeSql (sql) / / second way: declare a set of operations to execute sql StatementSet stmtSet = tEnv.createStatementSet (); stmtSet.addInsertSql (sql); TableResult tableResult = stmtSet.execute (); tableResult.print ();}}

Print the result

+-+ | default_catalog.default_database.my_users | +-+ | -1 | +-+ 1 row in set here The study on "how to insert data into MYSQL database tables using Tbale SQL and Flink JDBC connectors" is over. I hope to be able to solve your doubts. The collocation of theory and practice can better help you learn, go and try it! If you want to continue to learn more related knowledge, please continue to follow the website, the editor will continue to work hard to bring you more practical articles!

Welcome to subscribe "Shulou Technology Information " to get latest news, interesting things and hot topics in the IT industry, and controls the hottest and latest Internet news, technology news and IT industry trends.