怎么從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)

這篇文章主要介紹“怎么從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)”，在日常操作中，相信很多人在怎么從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)問題上存在疑惑，小編查閱了各式資料，整理出簡單好用的操作方法，希望對大家解答”怎么從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)”的疑惑有所幫助！接下來，請跟著小編一起來學(xué)習(xí)吧！

使用Tbale SQL與Flink JDBC連接器從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)

示例環(huán)境

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

SelectToMysql.java

package com.flink.examples.mysql;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Description 使用Tbale&SQL與Flink JDBC連接器從MYSQL數(shù)據(jù)庫表中SELECT選取數(shù)據(jù)。
 */
public class SelectToMysql {
    /**
     官方參考：https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html

     分區(qū)掃描
     為了加速并行Source任務(wù)實(shí)例中的數(shù)據(jù)讀取，F(xiàn)link為JDBC表提供了分區(qū)掃描功能。
         scan.partition.column：用于對輸入進(jìn)行分區(qū)的列名。
         scan.partition.num：分區(qū)數(shù)。
         scan.partition.lower-bound：第一個(gè)分區(qū)的最小值。
         scan.partition.upper-bound：最后一個(gè)分區(qū)的最大值。
     */

    //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" +
            "  'connector.read.fetch-size' = '3' \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);
        //注冊mysql數(shù)據(jù)維表
        tEnv.executeSql(table_sql);

        String sql = "select id,name,age,status from my_users";
        //第一種：執(zhí)行SQL
        //Table table = tEnv.sqlQuery(sql);

        //第二種：通過方法拼裝執(zhí)行語句
        Table table = tEnv.from("my_users").select($("id"), $("name"), $("age"), $("status")).where($("status").isEqual(0));

        //打印字段結(jié)構(gòu)
        table.printSchema();

        //table 轉(zhuǎn)成 dataStream 流
        DataStream<Row> behaviorStream = tEnv.toAppendStream(table, Row.class);
        behaviorStream.flatMap(new FlatMapFunction<Row, Object>() {
            @Override
            public void flatMap(Row value, Collector<Object> out) throws Exception {
                System.out.println(value.toString());
                Thread.sleep(1 * 1000);
            }
        }).print();
        env.execute();
    }
}

  打印結(jié)果

root
 |-- id: BIGINT NOT NULL
 |-- name: STRING
 |-- age: INT
 |-- status: INT

1,nike,16,0
2,nike,18,0

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