Tableapi & SQL and MySQL insert data of Flink

Use Tbale&SQL And Flink JDBC Connector inserts data into MYSQL Database table

Sample environment

java.version: 1.8.x
flink.version: 1.11.1
kafka：2.11

Sample data source  （ Project code cloud download ）

Flink System examples And Build development environment and data

Sample module  （pom.xml）

Flink System examples And TableAPI & SQL And Sample module

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  Use Tbale&SQL And Flink JDBC Connector inserts data into MYSQL Database table 
 */
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 How to write it , All attribute names are in JdbcTableSourceSinkFactory The factory class defines 
    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', --  The 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 {
        // structure StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        // structure EnvironmentSettings  And designate Blink Planner
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        // structure StreamTableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
        // register mysql Data dimension table 
        tEnv.executeSql(table_sql);

        // perform SQL,id=0 It's because of id Fields are self incrementing primary keys , by 0 be mysql Recognition will default to self increment instead of 
        String sql = "insert into my_users (id,name,age,status) values(0,'tom',31,0)";

        //  The first way ： Direct execution sql
//        TableResult tableResult = tEnv.executeSql(sql);

        // The second way ： Declare a collection of operations to perform sql
        StatementSet stmtSet = tEnv.createStatementSet();
        stmtSet.addInsertSql(sql);
        TableResult tableResult = stmtSet.execute();

        tableResult.print();
    }
}

  Print the results

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