flink sql

数据类型

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
BatchTableEnvironment tableEnv = BatchTableEnvironment.create(env);

StreamExecutionEnvironment env =　StreamExecutionEnvironment.getExecutionEnvironment();
StreamTableEnvironment tEnv=StreamTableEnvironment.create(env)

flink sql

数据类型

DataSet<Order> orderA = env.fromCollection(Arrays.asList(
                new Order(1L, "beer", 3),
                new Order(3L, "rubber", 2),
                new Order(1L, "diaper", 4)
               ));

// register Orders table in table environment
        Table Orders = tEnv.fromDataSet(orderA,"user,product,amount");
//        tEnv.registerTable("Orders",tableA);

// specify table program
//    Table orders = tEnv.from("Orders"); // schema (a, b, c, rowtime)

Table counts = Orders
              .groupBy($("user"),$("product"))
                .select($("user"),$("product"),$("amount").sum().as("cnt"));

Flink SQL的table api只支持

only_full_group_by模式[1]

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)));

// convert DataStream to Table
Table tableA = tEnv.fromDataStream(orderA, $("user"), $("product"), $("amount"));
// register DataStream as Table
tEnv.createTemporaryView("OrderB", orderB, $("user"), $("product"), $("amount"));