這篇文章主要講解了“flinksql表的查詢轉(zhuǎn)換方法”，文中的講解內(nèi)容簡(jiǎn)單清晰，易于學(xué)習(xí)與理解，下面請(qǐng)大家跟著小編的思路慢慢深入，一起來(lái)研究和學(xué)習(xí)“flinksql表的查詢轉(zhuǎn)換方法”吧！

網(wǎng)站建設(shè)公司,為您提供網(wǎng)站建設(shè),網(wǎng)站制作,網(wǎng)頁(yè)設(shè)計(jì)及定制網(wǎng)站建設(shè)服務(wù),專注于企業(yè)網(wǎng)站制作,高端網(wǎng)頁(yè)制作,對(duì)成都服務(wù)器托管等多個(gè)行業(yè)擁有豐富的網(wǎng)站建設(shè)經(jīng)驗(yàn)的網(wǎng)站建設(shè)公司。專業(yè)網(wǎng)站設(shè)計(jì),網(wǎng)站優(yōu)化推廣哪家好,專業(yè)成都網(wǎng)站推廣優(yōu)化,H5建站,響應(yīng)式網(wǎng)站。

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、創(chuàng)建表執(zhí)行環(huán)節(jié)
        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())  // 定義表的結(jié)構(gòu)
                        .field("b", DataTypes.STRING())
                        .field("c", DataTypes.STRING())

                )
                .createTemporaryTable("xxx");

//        表的查詢與轉(zhuǎn)換
        Table xxx = tableEnv.from("xxx");
//        簡(jiǎn)單查詢
        Table select = xxx.select("a, b").filter($("a").isEqual("a"));
       Table select2 = select.groupBy($("a"))
                .select($("a"), $("a").count().as("count"));

        select2.printSchema();


//        可撤回的方式  輸出結(jié)果前面會(huì)有一列 boolean true 表示以這次為準(zhǔn)， false為作廢
        tableEnv.toRetractStream(select2,  Row.class ).print();



        env.execute("job");


    }
}

感謝各位的閱讀，以上就是“flinksql表的查詢轉(zhuǎn)換方法”的內(nèi)容了，經(jīng)過(guò)本文的學(xué)習(xí)后，相信大家對(duì)flinksql表的查詢轉(zhuǎn)換方法這一問(wèn)題有了更深刻的體會(huì)，具體使用情況還需要大家實(shí)踐驗(yàn)證。這里是創(chuàng)新互聯(lián)，小編將為大家推送更多相關(guān)知識(shí)點(diǎn)的文章，歡迎關(guān)注！