flink sql

數據類型

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