十年网站开发经验 + 多家企业客户 + 靠谱的建站团队
量身定制 + 运营维护+专业推广+无忧售后,网站问题一站解决
这篇文章主要介绍“怎么从MySQL数据库表中SELECT选取数据”,在日常操作中,相信很多人在怎么从MYSQL数据库表中SELECT选取数据问题上存在疑惑,小编查阅了各式资料,整理出简单好用的操作方法,希望对大家解答”怎么从MYSQL数据库表中SELECT选取数据”的疑惑有所帮助!接下来,请跟着小编一起来学习吧!
创新互联建站是一家专业提供金牛企业网站建设,专注与网站建设、成都做网站、H5技术、小程序制作等业务。10年已为金牛众多企业、政府机构等服务。创新互联专业网站制作公司优惠进行中。
使用Tbale SQL与Flink JDBC连接器从MYSQL数据库表中SELECT选取数据
示例环境
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数据库表中SELECT选取数据。 */ public class SelectToMysql { /** 官方参考:https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/jdbc.html 分区扫描 为了加速并行Source任务实例中的数据读取,Flink为JDBC表提供了分区扫描功能。 scan.partition.column:用于对输入进行分区的列名。 scan.partition.num:分区数。 scan.partition.lower-bound:第一个分区的最小值。 scan.partition.upper-bound:最后一个分区的最大值。 */ //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 { //构建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); String sql = "select id,name,age,status from my_users"; //第一种:执行SQL //Table table = tEnv.sqlQuery(sql); //第二种:通过方法拼装执行语句 Table table = tEnv.from("my_users").select($("id"), $("name"), $("age"), $("status")).where($("status").isEqual(0)); //打印字段结构 table.printSchema(); //table 转成 dataStream 流 DataStreambehaviorStream = tEnv.toAppendStream(table, Row.class); behaviorStream.flatMap(new FlatMapFunction
() { @Override public void flatMap(Row value, Collector
打印结果
root |-- id: BIGINT NOT NULL |-- name: STRING |-- age: INT |-- status: INT 1,nike,16,0 2,nike,18,0
到此,关于“怎么从MYSQL数据库表中SELECT选取数据”的学习就结束了,希望能够解决大家的疑惑。理论与实践的搭配能更好的帮助大家学习,快去试试吧!若想继续学习更多相关知识,请继续关注创新互联网站,小编会继续努力为大家带来更多实用的文章!