怎么使用Tbale SQL與Flink JDBC連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表

這篇文章主要介紹“怎么使用Tbale SQL與Flink JDBC連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表”，在日常操作中，相信很多人在怎么使用Tbale SQL與Flink JDBC連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表問題上存在疑惑，小編查閱了各式資料，整理出簡(jiǎn)單好用的操作方法，希望對(duì)大家解答”怎么使用Tbale SQL與Flink JDBC連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表”的疑惑有所幫助！接下來，請(qǐng)跟著小編一起來學(xué)習(xí)吧！

示例環(huán)境

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連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表
 */
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', -- 數(shù)據(jù)庫需建立 \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 {
        //構(gòu)建StreamExecutionEnvironment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //構(gòu)建EnvironmentSettings 并指定Blink Planner
        EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
        //構(gòu)建StreamTableEnvironment
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, bsSettings);
        //注冊(cè)mysql數(shù)據(jù)維表
        tEnv.executeSql(table_sql);

        //執(zhí)行SQL,id=0是因id字段為自增主鍵，為0則mysql識(shí)別會(huì)默認(rèn)自增值代替
        String sql = "insert into my_users (id,name,age,status) values(0,'tom',31,0)";

        // 第一種方式：直接執(zhí)行sql
//        TableResult tableResult = tEnv.executeSql(sql);

        //第二種方式：聲明一個(gè)操作集合來執(zhí)行sql
        StatementSet stmtSet = tEnv.createStatementSet();
        stmtSet.addInsertSql(sql);
        TableResult tableResult = stmtSet.execute();

        tableResult.print();
    }
}

 打印結(jié)果

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

到此，關(guān)于“怎么使用Tbale SQL與Flink JDBC連接器將數(shù)據(jù)插入MYSQL數(shù)據(jù)庫表”的學(xué)習(xí)就結(jié)束了，希望能夠解決大家的疑惑。理論與實(shí)踐的搭配能更好的幫助大家學(xué)習(xí)，快去試試吧！若想繼續(xù)學(xué)習(xí)更多相關(guān)知識(shí)，請(qǐng)繼續(xù)關(guān)注億速云網(wǎng)站，小編會(huì)繼續(xù)努力為大家?guī)砀鄬?shí)用的文章！