这篇文章主要讲解了“flinksql表的查询转换方法”,文中的讲解内容简单清晰,易于学习与理解,下面请大家跟着小编的思路慢慢深入,一起来研究和学习“flinksql表的查询转换方法”吧!
package com.jd.data;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.descriptors.FileSystem;
import org.apache.flink.table.descriptors.OldCsv;
import org.apache.flink.table.descriptors.Schema;
import org.apache.flink.types.Row;
import scala.Tuple3;
import static org.apache.flink.table.api.Expressions.$;
public class TableAipDemo03 {
public static void main(String[] args) throws Exception {
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
env.setParallelism(1);
// 1、创建表执行环节
StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
String path = "/Users/liuhaijing/Desktop/flinktestword/aaa.txt";
tableEnv.connect(new FileSystem().path(path))
.withFormat(new OldCsv()) // 定义格式化方法
.withSchema(new Schema().field("a", DataTypes.STRING()) // 定义表的结构
.field("b", DataTypes.STRING())
.field("c", DataTypes.STRING())
)
.createTemporaryTable("xxx");
// 表的查询与转换
Table xxx = tableEnv.from("xxx");
// 简单查询
Table select = xxx.select("a, b").filter($("a").isEqual("a"));
Table select2 = select.groupBy($("a"))
.select($("a"), $("a").count().as("count"));
select2.printSchema();
// 可撤回的方式 输出结果前面会有一列 boolean true 表示以这次为准, false为作废
tableEnv.toRetractStream(select2, Row.class ).print();
env.execute("job");
}
}
感谢各位的阅读,以上就是“flinksql表的查询转换方法”的内容了,经过本文的学习后,相信大家对flinksql表的查询转换方法这一问题有了更深刻的体会,具体使用情况还需要大家实践验证。这里是亿速云,小编将为大家推送更多相关知识点的文章,欢迎关注!
亿速云「云服务器」,即开即用、新一代英特尔至强铂金CPU、三副本存储NVMe SSD云盘,价格低至29元/月。点击查看>>
免责声明:本站发布的内容(图片、视频和文字)以原创、转载和分享为主,文章观点不代表本网站立场,如果涉及侵权请联系站长邮箱:is@yisu.com进行举报,并提供相关证据,一经查实,将立刻删除涉嫌侵权内容。
原文链接:https://my.oschina.net/captainliu/blog/4973325