flinksql表的查询转换方法

这篇文章主要讲解了“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表的查询转换方法这一问题有了更深刻的体会，具体使用情况还需要大家实践验证。这里是亿速云，小编将为大家推送更多相关知识点的文章，欢迎关注！