flink Table API 与SQL入门实战

流处理和批处理都可以用,是非常的方便!

导入依赖

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-table_2.11</artifactId><version>1.7.0</version>
</dependency>

测试案例

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.{StreamTableEnvironment, table2TableConversions}
import org.apache.flink.table.api.{Table, TableEnvironment}object TableAPITest {def main(args: Array[String]): Unit = {//创建运行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//返回Scala流org.apache.flink.streaming.api.scala.StreamExecutionEnvironment的org.apache.flink.table.api.scala.StreamTableEnvironment。val tableenv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(env)//读取文本文档val stream: DataStream[String] = env.readTextFile("a.txt")//导入隐式转换import org.apache.flink.api.scala._//处理从文本读取的数据,封装成User类型val stream1: DataStream[User] = stream.map(x => {val arr: Array[String] = x.split(",")User(arr(0).toInt, arr(1), arr(2), arr(3).toInt, arr(4).toLong)})//将给定的DataStream转换为Table。//Table的字段名称是自动从DataStream的类型派生的。//如果流中的数据类型是case class可以直接根据case class的结构生成tableval table: Table = tableenv.fromDataStream(stream1)//查询一下,比如,就查询id,和sex这两列val table1: Table = table.select("id,sex")//将给定的Table转换为指定类型的附加DataStream。//注意,泛型要写对,比如我们上面查询的id是Int类型,sex是String类型,那么toAppendStream这块的泛型要匹配val value1: DataStream[(Int, String)] = table1.toAppendStream[(Int, String)]//打印查询后的内容value1.print()env.execute()}
}case class User(id: Int, sex: String, name: String, age: Int, num: Long)

运行结果

5> (2,女)
13> (5,男)
15> (6,男)
8> (3,女)
3> (1,男)
10> (4,男)

用SQL风格查询

这种查询方式更简单些,举个例子,和上面的运行结果是一样的!

import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.table.api.scala.{StreamTableEnvironment, table2TableConversions}
import org.apache.flink.table.api.{Table, TableEnvironment}object SQLAPITest {def main(args: Array[String]): Unit = {//创建运行环境val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment//table的环境是包含在上面的执行环境中的,所以从中可以获取table的执行环境val tableenv: StreamTableEnvironment = TableEnvironment.getTableEnvironment(env)//读取文本文档val stream: DataStream[String] = env.readTextFile("a.txt")//导入隐式转换import org.apache.flink.api.scala._//处理从文本读取的数据,封装成User类型val stream1: DataStream[User] = stream.map(x => {val arr: Array[String] = x.split(",")User(arr(0).toInt, arr(1), arr(2), arr(3).toInt, arr(4).toLong)})val table: Table = tableenv.fromDataStream(stream1)//以sql语法的形式查询,没注册表,因为如果向下面这种调用方式的话,会自动注册val result: Table = tableenv.sqlQuery(s"select id,name FROM $table")//table转为stream之后方便打印输出val value: DataStream[(Int, String)] = result.toAppendStream[(Int, String)]//打印查询后的内容value.print()env.execute()}
}

或者也可以先注册表格,然后再查询

tableenv.registerTable("user",table)//user要带个反单引号,因为这个user是flink的关键字,//所以  ,如果表明是其他的名字,可能就不需要加
val result: Table = tableenv.sqlQuery("select id,name FROM `user`")

总结

开发 Table API 与SQL API,是为了进一步降低开发难度

flink Table API 与SQL入门实战相关推荐

Flink Table API和SQL（下）
传送门: Flink Table API和SQL(上)(基本API介绍+流处理表的特性) Flink Table API和SQL(中)(时间属性及窗口+聚合查询+联结查询) Flink Table A ...
第六课大数据技术之Fink1.13的实战学习-Table Api和SQL
第六课大数据技术之Fink1.13的实战学习-Table Api和SQL 文章目录第六课大数据技术之Fink1.13的实战学习-Table Api和SQL 第一节 Fink SQL快速上手 1. ...
2021年大数据Flink（三十）：Flink Table API SQL 介绍
目录 Table API & SQL 介绍为什么需要Table API & SQL Table API& SQL发展历程架构升级查询处理器的选 ...
Flink学习笔记（十一）Table API 和 SQL
文章目录 11. Table API 和 SQL 11.1 快速上手 11.1.1 需要依赖 11.1.2 示例 11.2 基本 API 11.2.1 程序架构 11.2.2 创建表环境 11.2.3 ...
Flink的Table API 与SQL的流处理
1 流处理与SQL的区别 Table API和SQL,本质上还是基于关系型表的操作方式:而关系型表.SQL本身,一般是有界的,更适合批处理的场景.所以在流处理的过程中,有一些特殊概念. SQL 流 ...
Flink的Table API 与SQL介绍及调用
1 概述 DataSetAPI和DateStreamAPI是基于整个Flink的运行时环境做操作处理的,Table API和SQL是在DateStreamAPI上又包了一层.对于新版本的Blin ...
Flink Table Api SQL 初体验,Blink的使用
概述 Flink具有Table API和SQL-用于统一流和批处理. Table API是用于Scala和Java的语言集成查询API,它允许以非常直观的方式组合来自关系运算符(例如选择,过滤和联接) ...
（十八）Flink Table API SQL 编程指南 Table API 和Datastream API 集成
文章目录 DataStream 和 Table 之间的转换依赖项和导入配置执行行为 datastream API table API 批处理运行时模式 Changelog统一处理(仅插入)流 ...
Flink Table API SQL编程指南（自定义Sources Sinks）
TableSource TableSource提供对存储在外部系统(数据库,键值存储,消息队列)或文件中的数据的访问的通用接口.在TableEnvironment中注册TableSource后,可以通 ...

flink Table API 与SQL入门实战

用SQL风格查询

总结

flink Table API 与SQL入门实战相关推荐

最新文章

热门文章