目錄

案例二

需求

代碼實現-SQL

代碼實現-Table

---

案例二

需求

使用SQL和Table兩種方式對DataStream中的單詞進行統計

代碼實現-SQL

package cn.it.sql;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import static org.apache.flink.table.api.Expressions.$;/*** Author lanson* Desc*/
public class FlinkSQL_Table_Demo02 {public static void main(String[] args) throws Exception {//1.準備環境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);//2.SourceDataStream<WC> input = env.fromElements(new WC("Hello", 1),new WC("World", 1),new WC("Hello", 1));//3.注冊表tEnv.createTemporaryView("WordCount", input, $("word"), $("frequency"));//4.執行查詢Table resultTable = tEnv.sqlQuery("SELECT word, SUM(frequency) as frequency FROM WordCount GROUP BY word");//5.輸出結果//toAppendStream doesn't support consuming update changes which is produced by node GroupAggregate//DataStream<WC> resultDS = tEnv.toAppendStream(resultTable, WC.class);DataStream<Tuple2<Boolean, WC>> resultDS = tEnv.toRetractStream(resultTable, WC.class);resultDS.print();env.execute();}@Data@NoArgsConstructor@AllArgsConstructorpublic static class WC {public String word;public long frequency;}
}

代碼實現-Table

package cn.it.sql;import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;import static org.apache.flink.table.api.Expressions.$;/*** Author lanson* Desc*/
public class FlinkSQL_Table_Demo03 {public static void main(String[] args) throws Exception {//1.準備環境StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);//2.SourceDataStream<WC> input = env.fromElements(new WC("Hello", 1),new WC("World", 1),new WC("Hello", 1));//3.注冊表Table table = tEnv.fromDataStream(input);//4.執行查詢Table resultTable = table.groupBy($("word")).select($("word"), $("frequency").sum().as("frequency")).filter($("frequency").isEqual(2));//5.輸出結果DataStream<Tuple2<Boolean, WC>> resultDS = tEnv.toRetractStream(resultTable, WC.class);resultDS.print();env.execute();}@Data@NoArgsConstructor@AllArgsConstructorpublic static class WC {public String word;public long frequency;}
}

總結

以上是生活随笔為你收集整理的2021年大数据Flink（三十五）：​​​​​​​Table与SQL ​​​​​​案例二的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。