flink重温笔记（十五）： flinkSQL 顶层 API ——实时数据流转化为SQL表的操作

2024-03-13 02:20
文章标签 sql 操作 笔记 实时 api 转化 flink 顶层 十五 database 数据流 重温 flinksql

本文主要是介绍flink重温笔记（十五）： flinkSQL 顶层 API ——实时数据流转化为SQL表的操作，希望对大家解决编程问题提供一定的参考价值，需要的开发者们随着小编来一起学习吧！

Flink学习笔记

前言：今天是学习 flink 的第 15 天啦！学习了 flinkSQL 基础入门，主要是解决大数据领域数据处理采用表的方式，而不是写复杂代码逻辑，学会了如何初始化环境，鹅湖将流数据转化为表数据，以及如何查询表数据，结合自己实验猜想和代码实践，总结了很多自己的理解和想法，希望和大家多多交流！

Tips："分享是快乐的源泉💧，在我的博客里，不仅有知识的海洋🌊，还有满满的正能量加持💪，快来和我一起分享这份快乐吧😊！

喜欢我的博客的话，记得点个红心❤️和小关小注哦！您的支持是我创作的动力！"

文章目录

  • Flink学习笔记
    • 一、FlinkSQL 入门
      • 1. 引入依赖
      • 2. 创建 TableEnvironment
        • 2.1 配置版本的流式查询（Flink-Streaming-Query）
        • 2.2 配置老版本的批处理环境（Flink-Batch-Query）
        • 2.3 配置新版本的流式查询（Blink-Streaming-Query）
        • 2.4 配置新版本的批处理环境（Blink-Batch-Query）
      • 3. 查询表
        • 3.1 导包操作
        • 3.2 Table API 调用模型
        • 3.3 SQL 查询模型
      • 4. 将 DataStream 转化为表

一、FlinkSQL 入门

1. 引入依赖

在 FlinkSQL 学习阶段，需要的 pom 文件如下所示：

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"><modelVersion>4.0.0</modelVersion><groupId>cn.itcast</groupId><artifactId>flinksql_pro</artifactId><version>1.0-SNAPSHOT</version><build><plugins><plugin><groupId>org.apache.maven.plugins</groupId><artifactId>maven-compiler-plugin</artifactId><configuration><source>8</source><target>8</target></configuration></plugin></plugins></build><properties><flink.version>1.13.1</flink.version><java.version>1.8</java.version><scala.binary.version>2.11</scala.binary.version><hadoop.version>2.7.5</hadoop.version><hbase.version>2.0.0</hbase.version><zkclient.version>0.8</zkclient.version><hive.version>2.1.1</hive.version><mysql.version>5.1.47</mysql.version></properties><!--仓库配置--><repositories><repository><id>nexus-aliyun</id><name>Nexus aliyun</name><url>http://maven.aliyun.com/nexus/content/groups/public</url></repository><repository><id>central_maven</id><name>central maven</name><url>https://repo1.maven.org/maven2</url></repository><repository><id>cloudera</id><url>https://repository.cloudera.com/artifactory/cloudera-repos/</url></repository></repositories><dependencies><!-- Apache Flink dependencies --><!-- These dependencies are provided, because they should not be packaged into the JAR file. --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><!--<scope>provided</scope>--></dependency><!-- https://mvnrepository.com/artifact/junit/junit --><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>4.13.2</version><scope>test</scope></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-streaming-scala_${scala.binary.version}</artifactId><version>${flink.version}</version><!--<scope>compile</scope>--><exclusions><exclusion><groupId>log4j</groupId><artifactId>*</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-clients --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-clients_${scala.binary.version}</artifactId><version>${flink.version}</version><exclusions><exclusion><groupId>log4j</groupId><artifactId>*</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><!-- https://mvnrepository.com/artifact/org.apache.flink/flink-table --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-planner-blink_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-runtime-blink_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-table-api-scala-bridge_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-java</artifactId><version>${flink.version}</version><exclusions><exclusion><groupId>log4j</groupId><artifactId>*</artifactId></exclusion><exclusion><groupId>org.slf4j</groupId><artifactId>slf4j-log4j12</artifactId></exclusion></exclusions></dependency><!--kafka--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-kafka_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-sql-connector-kafka_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.kafka</groupId><artifactId>kafka-clients</artifactId><version>2.3.0</version></dependency><!--es6--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch6_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!--link-jdbc--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-jdbc_2.11</artifactId><version>${flink.version}</version></dependency><!--flink-hbase--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-hbase-2.2_2.11</artifactId><version>${flink.version}</version></dependency><!--hadoop--><!-- https://mvnrepository.com/artifact/org.apache.hadoop/hadoop-common --><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-common</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-hadoop-compatibility_2.11</artifactId><version>${flink.version}</version></dependency><dependency><groupId>org.apache.hadoop</groupId><artifactId>hadoop-client</artifactId><version>${hadoop.version}</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-shaded-hadoop-2-uber</artifactId><version>2.7.5-10.0</version></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-csv</artifactId><version>${flink.version}</version></dependency><!--flink-hbase--><dependency><groupId>org.apache.flink</groupId><artifactId>flink-json</artifactId><version>${flink.version}</version><!--<scope>test</scope>--></dependency><dependency><groupId>org.apache.flink</groupId><artifactId>flink-runtime_${scala.binary.version}</artifactId><version>${flink.version}</version><!--<scope>test</scope>--></dependency><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-classic</artifactId><version>1.2.3</version><!--<scope>test</scope>--></dependency><!-- https://mvnrepository.com/artifact/ch.qos.logback/logback-core --><dependency><groupId>ch.qos.logback</groupId><artifactId>logback-core</artifactId><version>1.2.3</version></dependency><!-- json --><dependency><groupId>com.alibaba</groupId><artifactId>fastjson</artifactId><version>1.2.5</version></dependency><!-- On hive --><!-- Flink Dependency --><dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-hive_${scala.binary.version}</artifactId><version>${flink.version}</version></dependency><!-- Hive Dependency --><dependency><groupId>org.apache.hive</groupId><artifactId>hive-exec</artifactId><version>${hive.version}</version></dependency><dependency><groupId>org.projectlombok</groupId><artifactId>lombok</artifactId><version>1.18.2</version><scope>provided</scope></dependency><!-- mysql 连接驱动 --><dependency><groupId>mysql</groupId><artifactId>mysql-connector-java</artifactId><version>${mysql.version}</version></dependency><dependency><groupId>junit</groupId><artifactId>junit</artifactId><version>RELEASE</version><scope>compile</scope></dependency></dependencies>
</project>

2. 创建 TableEnvironment

概述：TableEnvironment 是 Table API 和 SQL 的核心概念

作用：

  • 1- 注册 catalog，并在其内部注册表
  • 2- 执行 SQL 查询
  • 3- 注册用户自定义函数
  • 4- 将 DataStream 转化为表
  • 5- 保存对 ExecutionEnvironment 和 StreamExecutionEnvironment 的引用
2.1 配置版本的流式查询（Flink-Streaming-Query）
// **********************// FLINK STREAMING QUERY// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;StreamExecutionEnvironment fsEnv = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings fsSettings = EnvironmentSettings.newInstance().useOldPlanner() // 使用老版本planner.inStreamingMode() // 流处理模式.build();StreamTableEnvironment fsTableEnv = StreamTableEnvironment.create(fsEnv, fsSettings);
// or TableEnvironment fsTableEnv = TableEnvironment.create(fsSettings);
2.2 配置老版本的批处理环境（Flink-Batch-Query）
// ******************// FLINK BATCH QUERY// ******************
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.BatchTableEnvironment;ExecutionEnvironment fbEnv = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment fbTableEnv = BatchTableEnvironment.create(fbEnv);
2.3 配置新版本的流式查询（Blink-Streaming-Query）

和老版本的区别在于：useBlinkPlanner()

// **********************// BLINK STREAMING QUERY// **********************
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;StreamExecutionEnvironment bsEnv = StreamExecutionEnvironment.getExecutionEnvironment();EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner()// 使用新版本planner.inStreamingMode()// 流处理模式.build();
StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(bsEnv, bsSettings);
// or TableEnvironment bsTableEnv = TableEnvironment.create(bsSettings);
2.4 配置新版本的批处理环境（Blink-Batch-Query）

和老版本的区别在于：

  • 老版本用的是BatchTableEnvironment，传入fbEnv

  • 新版本用的是TableEnvironment，传入bbSettings

// ******************// BLINK BATCH QUERY// ******************
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;EnvironmentSettings bbSettings = EnvironmentSettings.newInstance().useBlinkPlanner()// 使用新版本planner.inBatchMode()// 批处理模式.build();
TableEnvironment bbTableEnv = TableEnvironment.create(bbSettings);

3. 查询表

3.1 导包操作
import static org.apache.flink.table.api.Expressions.*;
3.2 Table API 调用模型
# 借助 table 环境，找到数据源表
Table orderTable = tableEnv.from("inputTable");# 调用 table API 进行查询
# select中，$（字段名）
# filter中，可以过滤操作Table resultTable = orderTable.select($("id"),$("timestamp"),$("category"),$("areaName"),$("money")).filter($("areaName").isEqual("北京"));# 如果有分组聚类的话，groupBy 需要写在 select 前面
Table aggResultSqlTable = orderTable.groupBy($("areaName")).select($("areaName"), $("id").count().as("cnt"));
3.3 SQL 查询模型
# 借助 table 环境，找到数据源表
Table orderTable = tableEnv.from("inputTable");# 借助 sqlQuery 方法进行 SQL 查询
Table resultTable2  = tableEnv.sqlQuery("select id,`timestamp`,category,areaName,money from inputTable where areaName='北京'");

4. 将 DataStream 转化为表

# 读取的数据文件可以放在 resource 目录下
String filePath = 所在的类名.class.getClassLoader.getResource（“文件名”）.getPath（）；# 读取数据 ->
env.readTextFile（）# map函数转化类型# 将数据流转化为表格
tableEnv.fromDataStream（dataStream）

案例：将 DataStream 转化为 表

数据源：order.csv，放在 resource 目录下

user_001,1621718199,10.1,电脑
user_001,1621718201,14.1,手机
user_002,1621718202,82.5,手机
user_001,1621718205,15.6,电脑
user_004,1621718207,10.2,家电
user_001,1621718208,15.8,电脑
user_005,1621718212,56.1,电脑
user_002,1621718260,40.3,家电
user_001,1621718580,11.5,家居
user_001,1621718860,61.6,家居

代码：

package cn.itcast.day01;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
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 static org.apache.flink.table.api.Expressions.$;/*** @author lql* @time 2024-03-12 14:34:40* @description TODO：将 DataStream 转化为表*/
public class DataStreamToTable {public static void main(String[] args) throws Exception {// todo 1) 初始化 table 环境// 1.1 流处理环境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();// 1.2 setting环境EnvironmentSettings bsSettings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();// 1.3 表环境StreamTableEnvironment bsTableEnv = StreamTableEnvironment.create(env, bsSettings);// todo 2) 用流环境读取数据源String filePath = DataStreamToTable.class.getClassLoader().getResource("order.csv").getPath();DataStreamSource<String> inputStream = env.readTextFile(filePath);// todo 3) map 成为样例数据类型SingleOutputStreamOperator<OrderInfo> dataStream = inputStream.map(new MapFunction<String, OrderInfo>() {@Overridepublic OrderInfo map(String data) throws Exception {String[] dataArray = data.split(",");return new OrderInfo(dataArray[0],dataArray[1],Double.parseDouble(dataArray[2]),dataArray[3]);}});// todo 4) 将数据流化成表Table dataTable = bsTableEnv.fromDataStream(dataStream);// todo 5) 读取表格数据// 方法一： 调用 api 获得数据Table resultTable = dataTable.select($("id"), $("timestamp"), $("money"),$("category")).filter($("category").isEqual("电脑"));// 将表转化成为流打印bsTableEnv.toAppendStream(resultTable, Row.class).print("方法一：调用api的结果");// 方法二：临时表，sql查询获得数据bsTableEnv.createTemporaryView("inputTable",dataTable);Table resultTable1 = bsTableEnv.sqlQuery("SELECT * FROM inputTable WHERE category = '电脑'");bsTableEnv.toAppendStream(resultTable1, Row.class).print("方法二：调用sql查询的结果");env.execute();}@Data@AllArgsConstructor@NoArgsConstructorpublic static class OrderInfo {private String id;private String timestamp;private Double money;private String category;}
}

结果：

方法一：调用api的结果:5> +I[user_001, 1621718205, 15.6, 电脑]
方法二：调用sql查询的结果:5> +I[user_001, 1621718205, 15.6, 电脑]
方法一：调用api的结果:3> +I[user_001, 1621718199, 10.1, 电脑]
方法一：调用api的结果:7> +I[user_001, 1621718208, 15.8, 电脑]
方法二：调用sql查询的结果:3> +I[user_001, 1621718199, 10.1, 电脑]
方法二：调用sql查询的结果:7> +I[user_001, 1621718208, 15.8, 电脑]
方法一：调用api的结果:7> +I[user_005, 1621718212, 56.1, 电脑]
方法二：调用sql查询的结果:7> +I[user_005, 1621718212, 56.1, 电脑]

总结：

  • 1- 没有设置并行度为 1，打印结果乱序
  • 2- 方法一：调用 api 方法，流转化为表写执行逻辑
  • 3- 方法二：sql 查询，流转化为表，建立临时视图
  • 4- 两种方法都需要转化为流才能打印出来！

这篇关于flink重温笔记（十五）： flinkSQL 顶层 API ——实时数据流转化为SQL表的操作的文章就介绍到这儿，希望我们推荐的文章对编程师们有所帮助！


http://www.chinasem.cn/article/803364。

相关文章

MySQL更新某个字段拼接固定字符串的实现

MySQL更新某个字段拼接固定字符串的实现

《MySQL更新某个字段拼接固定字符串的实现》在MySQL中,我们经常需要对数据库中的某个字段进行更新操作,本文就来介绍一下MySQL更新某个字段拼接固定字符串的实现,感兴趣的可以了解一下... 目录1. 查看字段当前值2. 更新字段拼接固定字符串3. 验证更新结果mysql更新某个字段拼接固定字符串 -
阅读更多...
python连接本地SQL server详细图文教程

python连接本地SQL server详细图文教程

《python连接本地SQLserver详细图文教程》在数据分析领域,经常需要从数据库中获取数据进行分析和处理,下面：本文主要介绍python连接本地SQLserver的相关资料,文中通过代码... 目录一.设置本地账号1.新建用户2.开启双重验证3,开启TCP/IP本地服务二js.python连接实例1.
阅读更多...
Spring Boot项目中结合MyBatis实现MySQL的自动主从切换功能

Spring Boot项目中结合MyBatis实现MySQL的自动主从切换功能

《SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能》：本文主要介绍SpringBoot项目中结合MyBatis实现MySQL的自动主从切换功能,本文分步骤给大家介绍的... 目录原理解析1. mysql主从复制（Master-Slave Replication）2. 读写分离3.
阅读更多...
Ubuntu中远程连接Mysql数据库的详细图文教程

Ubuntu中远程连接Mysql数据库的详细图文教程

《Ubuntu中远程连接Mysql数据库的详细图文教程》Ubuntu是一个以桌面应用为主的Linux发行版操作系统,这篇文章主要为大家详细介绍了Ubuntu中远程连接Mysql数据库的详细图文教程,有... 目录1、版本2、检查有没有mysql2.1 查询是否安装了Mysql包2.2 查看Mysql版本2.
阅读更多...
基于SpringBoot+Mybatis实现Mysql分表

基于SpringBoot+Mybatis实现Mysql分表

《基于SpringBoot+Mybatis实现Mysql分表》这篇文章主要为大家详细介绍了基于SpringBoot+Mybatis实现Mysql分表的相关知识,文中的示例代码讲解详细,感兴趣的小伙伴可... 目录基本思路定义注解创建ThreadLocal创建拦截器业务处理基本思路1.根据创建时间字段按年进
阅读更多...
Python3.6连接MySQL的详细步骤

Python3.6连接MySQL的详细步骤

《Python3.6连接MySQL的详细步骤》在现代Web开发和数据处理中,Python与数据库的交互是必不可少的一部分,MySQL作为最流行的开源关系型数据库管理系统之一,与Python的结合可以实... 目录环境准备安装python 3.6安装mysql安装pymysql库连接到MySQL建立连接执行S
阅读更多...
MySQL双主搭建+keepalived高可用的实现

MySQL双主搭建+keepalived高可用的实现

《MySQL双主搭建+keepalived高可用的实现》本文主要介绍了MySQL双主搭建+keepalived高可用的实现,文中通过示例代码介绍的非常详细,对大家的学习或者工作具有一定的参考学习价值,... 目录一、测试环境准备二、主从搭建1.创建复制用户2.创建复制关系3.开启复制，确认复制是否成功4.同
阅读更多...
MyBatis 动态 SQL 优化之标签的实战与技巧(常见用法)

MyBatis 动态 SQL 优化之标签的实战与技巧(常见用法)

《MyBatis动态SQL优化之标签的实战与技巧(常见用法)》本文通过详细的示例和实际应用场景,介绍了如何有效利用这些标签来优化MyBatis配置,提升开发效率,确保SQL的高效执行和安全性,感... 目录动态SQL详解一、动态SQL的核心概念1.1 什么是动态SQL？1.2 动态SQL的优点1.3 动态S
阅读更多...
Mysql表的简单操作(基本技能)

Mysql表的简单操作(基本技能)

《Mysql表的简单操作(基本技能)》在数据库中,表的操作主要包括表的创建、查看、修改、删除等,了解如何操作这些表是数据库管理和开发的基本技能,本文给大家介绍Mysql表的简单操作,感兴趣的朋友一起看... 目录3.1 创建表 3.2 查看表结构3.3 修改表3.4 实践案例：修改表在数据库中，表的操作主要
阅读更多...
C# WinForms存储过程操作数据库的实例讲解

C# WinForms存储过程操作数据库的实例讲解

《C#WinForms存储过程操作数据库的实例讲解》：本文主要介绍C#WinForms存储过程操作数据库的实例,具有很好的参考价值,希望对大家有所帮助,如有错误或未考虑完全的地方,望不吝赐教... 目录一、存储过程基础二、C# 调用流程1. 数据库连接配置2. 执行存储过程（增删改）3. 查询数据三、事务处
阅读更多...

Copyright China编程 23002807@qq.com

沪ICP备2023033072号-2