​​​​​​

目录

​​​​​​案例一

需求

代码实现

案例一

需求

将DataStream注册为Table和View并进行SQL统计

代码实现

  

   

    

     

    
    

     

      package cn.itcast.sql;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      import lombok.AllArgsConstructor;
     
    
   

    

     

    
    

     

      import lombok.Data;
     
    
   

    

     

    
    

     

      import lombok.NoArgsConstructor;
     
    
   

    

     

    
    

     

      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 java.util.Arrays;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      import static org.apache.flink.table.api.Expressions.$;
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

      /**
     
    
   

    

     

    
    

     

       * Author itcast
     
    
   

    

     

    
    

     

       * Desc
     
    
   

    

     

    
    

     

       */
     
    
   

    

     

    
    

     

      public class FlinkSQL_Table_Demo01 {
     
    
   

    

     

    
    

     

          public static void main(String[] args) throws Exception {
     
    
   

    

     

    
    

     

              //1.准备环境
     
    
   

    

     

    
    

     

              StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
     
    
   

    

     

    
    

     

              //EnvironmentSettings settings = EnvironmentSettings.newInstance().useBlinkPlanner().inStreamingMode().build();
     
    
   

    

     

    
    

     

              //StreamTableEnvironment tEnv = StreamTableEnvironment.create(env, settings);
     
    
   

    

     

    
    

     

              StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              //2.Source
     
    
   

    

     

    
    

     

              DataStream<Order> orderA = env.fromCollection(Arrays.asList(
     
    
   

    

     

    
    

     

                      new Order(1L, "beer", 3),
     
    
   

    

     

    
    

     

                      new Order(1L, "diaper", 4),
     
    
   

    

     

    
    

     

                      new Order(3L, "rubber", 2)));
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              DataStream<Order> orderB = env.fromCollection(Arrays.asList(
     
    
   

    

     

    
    

     

                      new Order(2L, "pen", 3),
     
    
   

    

     

    
    

     

                      new Order(2L, "rubber", 3),
     
    
   

    

     

    
    

     

                      new Order(4L, "beer", 1)));
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              //3.注册表
     
    
   

    

     

    
    

     

              // convert DataStream to Table
     
    
   

    

     

    
    

     

              Table tableA = tEnv.fromDataStream(orderA, $("user"), $("product"), $("amount"));
     
    
   

    

     

    
    

     

              // register DataStream as Table
     
    
   

    

     

    
    

     

              tEnv.createTemporaryView("OrderB", orderB, $("user"), $("product"), $("amount"));
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              //4.执行查询
     
    
   

    

     

    
    

     

              System.out.println(tableA);
     
    
   

    

     

    
    

     

              // union the two tables
     
    
   

    

     

    
    

     

              Table resultTable = tEnv.sqlQuery(
     
    
   

    

     

    
    

     

                      "SELECT * FROM " + tableA + " WHERE amount > 2 " +
     
    
   

    

     

    
    

     

                      "UNION ALL " +
     
    
   

    

     

    
    

     

                      "SELECT * FROM OrderB WHERE amount < 2"
     
    
   

    

     

    
    

     

              );
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              //5.输出结果
     
    
   

    

     

    
    

     

              DataStream<Order> resultDS = tEnv.toAppendStream(resultTable, Order.class);
     
    
   

    

     

    
    

     

              resultDS.print();
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

              env.execute();
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     
 
     
    
   

    

     

    
    

     

          @Data
     
    
   

    

     

    
    

     

          @NoArgsConstructor
     
    
   

    

     

    
    

     

          @AllArgsConstructor
     
    
   

    

     

    
    

     

          public static class Order {
     
    
   

    

     

    
    

     

              public Long user;
     
    
   

    

     

    
    

     

              public String product;
     
    
   

    

     

    
    

     

              public int amount;
     
    
   

    

     

    
    

     

          }
     
    
   

    

     

    
    

     

      }
     
    
  
 

文章来源: lansonli.blog.csdn.net，作者：Lansonli，版权归原作者所有，如需转载，请联系作者。

原文链接：lansonli.blog.csdn.net/article/details/116357228