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表的查詢轉換方法這一問題有了更深刻的體會，具體使用情況還需要大家實踐驗證。這里是億速云，小編將為大家推送更多相關知識點的文章，歡迎關注！