17、Flink 之Table API_ Table API 支持的操作（2）

标签：rowtime 17 id API Table DataTypes true tenv

(文章目录)

本文通过示例介绍了如何使用table api进行表的联接、排序、insert、group window、over window、以及基于行的操作，每个示例都是经过验证的、可运行的，并将运行结果展示在输出后面。关于如何使用table api进行表、视图、窗口函数的操作，同时也介绍了table api对表的查询、过滤、列、聚合以及join操作详见上篇文章：17、Flink 之Table API: Table API 支持的操作（1）。本文依赖flink、kafka、hive集群能正常使用。本文示例java api的实现是通过Flink 1.17版本做的示例，SQL是在Flink 1.17版本的环境中运行的。本文分为6个部分，即表的联接、表的排序、表的insert、group window、over window、基于行的表操作。

一、Table API介绍

本文的所有示例需要依赖的maven见本篇的上一篇：17、Flink 之Table API: Table API 支持的操作（1）

6、表的union、unionall、intersect、intersectall、minus、minusall和in的操作

本示例的运行结果均在执行用例中，其中用例只能在批模式下工作，用例特意说明了，如果没说明的则意味着流批模式均可。

import java.util.Arrays;
import java.util.List;

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.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Executable;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.tablesql.TestTableAPIJoinOperationDemo.Order;
import org.tablesql.TestTableAPIJoinOperationDemo.User;

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

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestTableAPIJoinOperationDemo2 {

	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private double balance;
		private Long rowtime;
	}
	
	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class Order {
		private long id;
		private long user_id;
		private double amount;
		private Long rowtime;
	}

	final static List<User> userList = Arrays.asList(
			new User(1L, "alan", 18, 1698742358391L), 
			new User(2L, "alan", 19, 1698742359396L), 
			new User(3L, "alan", 25, 1698742360407L),
			new User(4L, "alanchan", 28, 1698742361409L), 
			new User(5L, "alanchan", 29, 1698742362424L)
			);
	
	final static List<Order> orderList = Arrays.asList(
			new Order(1L, 1, 18, 1698742358391L), 
			new Order(2L, 2, 19, 1698742359396L), 
			new Order(3L, 1, 25, 1698742360407L),
			new Order(4L, 3, 28, 1698742361409L), 
			new Order(5L, 1, 29, 1698742362424L),
			new Order(6L, 4, 49, 1698742362424L)
			);
	
	 // 创建输出表
	final static String sinkSql = "CREATE TABLE sink_table (\n" +
            "  id BIGINT,\n" +
            "  user_id BIGINT,\n" +
            "  amount DOUBLE,\n" +
            "  rowtime BIGINT\n" +
            ") WITH (\n" +
            "  'connector' = 'print'\n" +
            ")";
	
	/**
	 * 
	 * @throws Exception
	 */
	static void testUnionBySQL() throws Exception {
		// TODO 0.env
				StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
				EnvironmentSettings settings = EnvironmentSettings.newInstance().inStreamingMode().build();
				StreamTableEnvironment tenv = StreamTableEnvironment.create(env, settings);

				DataStream<Order> orderA = env.fromCollection(orderList);
				DataStream<Order> orderB = env.fromCollection(orderList);

				// 将DataStream数据转Table和View,然后查询
				Table tableA = tenv.fromDataStream(orderA, $("id"), $("user_id"), $("amount"),$("rowtime"));
				tenv.createTemporaryView("tableB", orderB, $("id"), $("user_id"), $("amount"),$("rowtime"));

				// 查询:tableA中amount>2的和tableB中amount>1的数据最后合并
//				select * from tableA where amount > 2
//				union
//				 select * from tableB where amount > 1
				String sql = "select * from " + tableA + " where amount > 2  union   select * from tableB where amount > 1";

				Table resultTable = tenv.sqlQuery(sql);

				DataStream<Tuple2<Boolean, Order>> resultDS = tenv.toRetractStream(resultTable, Order.class);// union使用toRetractStream
//				String sql = "select * from " + tableA + " where amount > 2  union   select * from tableB where amount > 1";
//				9> (true,TestTableAPIJoinOperationDemo2.Order(id=1, user_id=1, amount=18.0, rowtime=1698742358391))
//				8> (true,TestTableAPIJoinOperationDemo2.Order(id=2, user_id=2, amount=19.0, rowtime=1698742359396))
//				4> (true,TestTableAPIJoinOperationDemo2.Order(id=5, user_id=1, amount=29.0, rowtime=1698742362424))
//				8> (true,TestTableAPIJoinOperationDemo2.Order(id=4, user_id=3, amount=28.0, rowtime=1698742361409))
//				14> (true,TestTableAPIJoinOperationDemo2.Order(id=6, user_id=4, amount=49.0, rowtime=1698742362424))
//				6> (true,TestTableAPIJoinOperationDemo2.Order(id=3, user_id=1, amount=25.0, rowtime=1698742360407))
				
				// toAppendStream → 将计算后的数据append到结果DataStream中去
				// toRetractStream → 将计算后的新的数据在DataStream原数据的基础上更新true或是删除false
				// 类似StructuredStreaming中的append/update/complete

				// TODO 3.sink
				resultDS.print();

				// TODO 4.execute
				env.execute();
	}
	
	/**
	 * 和 SQL UNION 子句类似。Union 两张表会删除重复记录。两张表必须具有相同的字段类型。
	 * 本示例仅仅使用同一个表来演示
	 * 该操作只能是在批处理模式下
	 * 
	 * @throws Exception
	 */
	static void testUnion() throws Exception {
//		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
        
		Table ordersTable = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(5L, 1, 29, 1698742362424L),
						row(6L, 4, 49, 1698742362424L)
						));
				
		Table left = ordersTable.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table unionResult = left.union(left);
		
		tenv.createTemporaryView("order_union_t", unionResult);

		Table result = tenv.sqlQuery("select * from order_union_t");
		
		// 下面不能转换，只有流式表可以转成流
		// 出现异常：The UNION operation on two unbounded tables is currently not supported.
		//		DataStream<Tuple2<Boolean, Order>> resultDS = tenv.toRetractStream(result, Order.class);
		//		resultDS.print();
		
		//输出表
		tenv.executeSql(sinkSql);
		
		result.executeInsert("sink_table");
//				+I[6, 4, 49.0, 1698742362424]
//				+I[5, 1, 29.0, 1698742362424]
//				+I[1, 1, 18.0, 1698742358391]
//				+I[3, 1, 25.0, 1698742360407]
//				+I[4, 3, 28.0, 1698742361409]
//				+I[2, 2, 19.0, 1698742359396]

	}
	
	/**
	 * 和 SQL UNION ALL 子句类似。Union 两张表。 两张表必须具有相同的字段类型。
	 * 本示例仅仅使用同一个表来演示
	 * 
	 * @throws Exception
	 */
	static void testUnionAll() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"),$("balance"),$("rowtime"));
				
		Table left = usersTable.select($("id").as("userId"), $("name"), $("balance"),$("rowtime").as("u_rowtime"));
		
		Table result = left.unionAll(left);
		
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		14> (true,+I[5, alanchan, 29.0, 1698742362424])
//		8> (true,+I[4, alanchan, 28.0, 1698742361409])
//		5> (true,+I[1, alan, 18.0, 1698742358391])
//		10> (true,+I[1, alan, 18.0, 1698742358391])
//		11> (true,+I[2, alan, 19.0, 1698742359396])
//		6> (true,+I[2, alan, 19.0, 1698742359396])
//		7> (true,+I[3, alan, 25.0, 1698742360407])
//		13> (true,+I[4, alanchan, 28.0, 1698742361409])
//		12> (true,+I[3, alan, 25.0, 1698742360407])
//		9> (true,+I[5, alanchan, 29.0, 1698742362424])
		
		env.execute();
	}
	
	/**
	 * 和 SQL INTERSECT 子句类似。Intersect 返回两个表中都存在的记录。
	 * 如果一条记录在一张或两张表中存在多次，则只返回一条记录，也就是说，结果表中不存在重复的记录。
	 * 两张表必须具有相同的字段类型。
	 * 该操作只能是在批处理模式下
	 * 
	 * @throws Exception
	 */
	static void testIntersect() throws Exception  {
		EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
        
		Table ordersTableA = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(6L, 4, 49, 1698742362424L)
						));
		
		Table ordersTableB = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(7L, 8, 4009, 1698782362424L)
						));
				
		Table left = ordersTableA.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		Table right = ordersTableB.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table intersectResult = left.intersect(right);
		
		tenv.createTemporaryView("order_intersect_t", intersectResult);

		Table result = tenv.sqlQuery("select * from order_intersect_t");
				
		//输出表
		tenv.executeSql(sinkSql);
		
		result.executeInsert("sink_table");
//		+I[1, 1, 18.0, 1698742358391]
		
	}
	
	/**
	 * 和 SQL INTERSECT ALL 子句类似。
	 * IntersectAll 返回两个表中都存在的记录。如果一条记录在两张表中出现多次，那么该记录返回的次数同该记录在两个表中都出现的次数一致，也就是说，结果表可能存在重复记录。
	 * 两张表必须具有相同的字段类型。
	 * 该操作只能是在批处理模式下
	 * 
	 * @throws Exception
	 */
	static void testIntersectAll() throws Exception  {
		EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
        
		Table ordersTableA = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(6L, 4, 49, 1698742362424L)
						));
		
		Table ordersTableB = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(7L, 8, 4009, 1698782362424L)
						));
				
		Table left = ordersTableA.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		Table right = ordersTableB.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table intersectResult = left.intersectAll(right);
		
		tenv.createTemporaryView("order_intersect_t", intersectResult);

		Table result = tenv.sqlQuery("select * from order_intersect_t");
				
		//输出表
		tenv.executeSql(sinkSql);
		
		result.executeInsert("sink_table");
//				+I[2, 2, 19.0, 1698742359396]
//				+I[1, 1, 18.0, 1698742358391]
		
	}
	
	/**
	 * 和 SQL EXCEPT 子句类似。Minus 返回左表中存在且右表中不存在的记录。
	 * 左表中的重复记录只返回一次，换句话说，结果表中没有重复记录。
	 * 两张表必须具有相同的字段类型。
	 * 该操作只能是在批处理模式下
	 * 
	 * @throws Exception
	 */
	static void testMinus() throws Exception  {
		EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
        
		Table ordersTableA = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(6L, 4, 49, 1698742362424L)
						));
		
		Table ordersTableB = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(7L, 8, 4009, 1698782362424L)
						));
				
		Table left = ordersTableA.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		Table right = ordersTableB.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table intersectResult = left.minus(right);
		
		tenv.createTemporaryView("order_intersect_t", intersectResult);

		Table result = tenv.sqlQuery("select * from order_intersect_t");
				
		//输出表
		tenv.executeSql(sinkSql);
		
		result.executeInsert("sink_table");
//		+I[6, 4, 49.0, 1698742362424]
		
	}
	
	/**
	 * 和 SQL EXCEPT ALL 子句类似。
	 * MinusAll 返回右表中不存在的记录。在左表中出现 n 次且在右表中出现 m 次的记录，在结果表中出现 (n - m) 次，
	 * 例如，也就是说结果中删掉了在右表中存在重复记录的条数的记录。
	 * 两张表必须具有相同的字段类型。
	 * 该操作只能是在批处理模式下
	 * 
	 * @throws Exception
	 */
	static void testMinusAll() throws Exception  {
		EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
        
		Table ordersTableA = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(6L, 4, 49, 1698742362424L)
						));
		
		Table ordersTableB = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(7L, 8, 4009, 1698782362424L)
						));
				
		Table left = ordersTableA.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		Table right = ordersTableB.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table intersectResult = left.minus(right);
		
		tenv.createTemporaryView("order_intersect_t", intersectResult);

		Table result = tenv.sqlQuery("select * from order_intersect_t");
				
		//输出表
		tenv.executeSql(sinkSql);
		
		result.executeInsert("sink_table");
//		+I[6, 4, 49.0, 1698742362424]
		
	}
	
	/**
	 * 和 SQL IN 子句类似。如果表达式的值存在于给定表的子查询中，那么 In 子句返回 true。
	 * 子查询表必须由一列组成。
	 * 这个列必须与表达式具有相同的数据类型。
	 * 
	 * @throws Exception
	 */
	static void testIn() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"),$("balance"),$("rowtime"));
		
		DataStream<Order> orders = env.fromCollection(orderList);
		Table ordersTable = tenv.fromDataStream(orders, $("id"), $("user_id"), $("amount"),$("rowtime"));
		
		Table left = usersTable.select($("id").as("userId"), $("name"), $("balance"),$("rowtime").as("u_rowtime"));
		Table right = ordersTable.select($("user_id"));
		
		Table result = left.select($("userId"), $("name"), $("balance"),$("u_rowtime")).where($("userId").in(right));

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		3> (true,+I[4, alanchan, 28.0, 1698742361409])
//		12> (true,+I[1, alan, 18.0, 1698742358391])
//		15> (true,+I[3, alan, 25.0, 1698742360407])
//		12> (true,+I[2, alan, 19.0, 1698742359396])
		env.execute();
	}
	
	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
//		testUnion();
//		testUnionAll();
//		testUnionBySQL();
//		testIntersect();
//		testIntersectAll() ;
//		testMinus();
//		testMinusAll();
		testIn();
		
	}

}

7、表的OrderBy, Offset 和 Fetch操作

在批处理模式下，也即有界情况下，order by 可以单独使用，排序也可以是任意字段，与一般数据库的排序结果一样。在流模式下，也即无界的情况下，order by需要和fetch一起使用，排序字段需要有时间属性，与一般数据库的排序有点差异。

需要说明的是order by 和offset&fetch都可以在批处理模式和流模式下工作。

Order By，和 SQL ORDER BY 子句类似。返回跨所有并行分区的全局有序记录。对于无界表，该操作需要对时间属性进行排序或进行后续的 fetch 操作。
Offset & Fetch，和 SQL 的 OFFSET 和 FETCH 子句类似。Offset 操作根据偏移位置来限定（可能是已排序的）结果集。Fetch 操作将（可能已排序的）结果集限制为前 n 行。通常，这两个操作前面都有一个排序操作。对于无界表，offset 操作需要 fetch 操作。

具体结果见下面示例

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

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.DataTypes;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.tablesql.TestTableAPIJoinOperationDemo2.Order;
import org.tablesql.TestTableAPIJoinOperationDemo2.User;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestTableAPIJoinOperationDemo3 {
	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private double balance;
		private Long rowtime;
	}
	
	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class Order {
		private long id;
		private long user_id;
		private double amount;
		private Long rowtime;
	}

	final static List<User> userList = Arrays.asList(
			new User(1L, "alan", 18, 1698742358391L), 
			new User(2L, "alan", 19, 1698742359396L), 
			new User(3L, "alan", 25, 1698742360407L),
			new User(4L, "alanchan", 28, 1698742361409L), 
			new User(5L, "alanchan", 29, 1698742362424L)
			);
	
	final static List<Order> orderList = Arrays.asList(
			new Order(1L, 1, 18, 1698742358391L), 
			new Order(2L, 2, 19, 1698742359396L), 
			new Order(3L, 1, 25, 1698742360407L),
			new Order(4L, 3, 28, 1698742361409L), 
			new Order(5L, 1, 29, 1698742362424L),
			new Order(6L, 4, 49, 1698742362424L)
			);
	
	 // 创建输出表
	final static String sinkSql = "CREATE TABLE sink_table (\n" +
            "  id BIGINT,\n" +
            "  user_id BIGINT,\n" +
            "  amount DOUBLE,\n" +
            "  rowtime BIGINT\n" +
            ") WITH (\n" +
            "  'connector' = 'print'\n" +
            ")";
	
	/**
	 * Order By
	 * 和 SQL ORDER BY 子句类似。返回跨所有并行分区的全局有序记录。
	 * 对于无界表，该操作需要对时间属性进行排序或进行后续的 fetch 操作。
	 * Sort on a non-time-attribute field is not supported.
	 * 
	 * Offset & Fetch
	 * 和 SQL 的 OFFSET 和 FETCH 子句类似。
	 * Offset 操作根据偏移位置来限定（可能是已排序的）结果集。
	 * Fetch 操作将（可能已排序的）结果集限制为前 n 行。
	 * 通常，这两个操作前面都有一个排序操作。对于无界表，offset 操作需要 fetch 操作。
	 * 
	 * @throws Exception
	 */
	static void testOrderByWithUnbounded() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		env.setParallelism(1);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		usersTable.printSchema();
		
		// 从已排序的结果集中返回前3条记录
		Table result = usersTable.orderBy($("rowtime").desc()).fetch(3);

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
//		resultDS.print();
//		(true,+I[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(true,+I[2, alan, 19.0, 2023-10-31T08:52:39.396])
//		(true,+I[3, alan, 25.0, 2023-10-31T08:52:40.407])
//		(false,-D[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(true,+I[4, alanchan, 28.0, 2023-10-31T08:52:41.409])
//		(false,-D[2, alan, 19.0, 2023-10-31T08:52:39.396])
//		(true,+I[5, alanchan, 29.0, 2023-10-31T08:52:42.424])

		// 从已排序的结果集中返回跳过2条记录之后的所有记录
		Table result2 = usersTable.orderBy($("rowtime").desc()).offset(2).fetch(4);

		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
		result2DS.print();
//		(true,+I[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(false,-U[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(true,+U[2, alan, 19.0, 2023-10-31T08:52:39.396])
//		(true,+I[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(false,-U[2, alan, 19.0, 2023-10-31T08:52:39.396])
//		(true,+U[3, alan, 25.0, 2023-10-31T08:52:40.407])
//		(false,-U[1, alan, 18.0, 2023-10-31T08:52:38.391])
//		(true,+U[2, alan, 19.0, 2023-10-31T08:52:39.396])
//		(true,+I[1, alan, 18.0, 2023-10-31T08:52:38.391])
		
		env.execute();
	}
	
	/**
	 * 和 SQL ORDER BY 子句类似。返回跨所有并行分区的全局有序记录。
	 * 对于无界表，该操作需要对时间属性进行排序或进行后续的 fetch 操作。
	 * 这个和一般的查询数据库的结果比较类似
	 * 
	 * @throws Exception
	 */
	static void testOrderByWithBounded() throws Exception {
        EnvironmentSettings env = EnvironmentSettings.newInstance().inBatchMode() .build();
        TableEnvironment tenv = TableEnvironment.create(env);
       
		Table ordersTable = tenv.fromValues(
				DataTypes.ROW(
						DataTypes.FIELD("id", DataTypes.BIGINT()),
						DataTypes.FIELD("user_id", DataTypes.BIGINT()),
						DataTypes.FIELD("amount", DataTypes.BIGINT()),
						DataTypes.FIELD("rowtime", DataTypes.BIGINT())
						),
				Arrays.asList(
						row(1L, 1, 18, 1698742358391L), 
						row(2L, 2, 19, 1698742359396L), 
						row(3L, 1, 25, 1698742360407L),
						row(4L, 3, 28, 1698742361409L), 
						row(5L, 1, 29, 1698742362424L),
						row(6L, 4, 49, 1698742362424L)
						));
		
		Table left = ordersTable.select($("id"), $("user_id"),$("amount"),$("rowtime"));
		
		Table orderByResult = left.orderBy($("amount").desc());
		
		tenv.createTemporaryView("order_union_t", orderByResult);

		Table result = tenv.sqlQuery("select * from order_union_t");
		
		//输出表
		tenv.executeSql(sinkSql);
//				+I[6, 4, 49.0, 1698742362424]
//				+I[5, 1, 29.0, 1698742362424]
//				+I[4, 3, 28.0, 1698742361409]
//				+I[3, 1, 25.0, 1698742360407]
//				+I[2, 2, 19.0, 1698742359396]
//				+I[1, 1, 18.0, 1698742358391]

		result.executeInsert("sink_table");
	}
	
	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
//		testOrderByWithUnbounded();
		testOrderByWithBounded();
	}

}

8、表的insert操作

和 SQL 查询中的 INSERT INTO 子句类似，该方法执行对已注册的输出表的插入操作。 insertInto() 方法会将 INSERT INTO 转换为一个 TablePipeline。该数据流可以用 TablePipeline.explain() 来解释，用 TablePipeline.execute() 来执行。

输出表必须已注册在 TableEnvironment中。此外，已注册表的 schema 必须与查询中的 schema 相匹配。

该示例仅仅展示一个方法，运行环境和其他的示例一致，并且本示例仅仅展示的是insertInto，也可以使用executeInsert方法，在其他示例中有展示其使用。

static void testInsert() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<Order> orderA = env.fromCollection(orderList);
		DataStream<Order> orderB = env.fromCollection(
				Arrays.asList(
						new Order(10L, 1, 18, 1698742358391L), 
						new Order(16L, 4, 49, 1698742362424L)
						)
				);
		
		Table tableA = tenv.fromDataStream(orderA, $("id"), $("user_id"), $("amount"),$("rowtime"));
		Table tableB = tenv.fromDataStream(orderB, $("id"), $("user_id"), $("amount"),$("rowtime"));
		tenv.executeSql(sinkSql);
		
		tableA.insertInto("sink_table").execute();
		tableB.insertInto("sink_table").execute();
//				+I[1, 1, 18.0, 1698742358391]
//				+I[2, 2, 19.0, 1698742359396]
//				+I[3, 1, 25.0, 1698742360407]
//				+I[4, 3, 28.0, 1698742361409]
//				+I[5, 1, 29.0, 1698742362424]
//				+I[6, 4, 49.0, 1698742362424]
//				+I[10, 1, 18.0, 1698742358391]
//				+I[16, 4, 49.0, 1698742362424]
		
	}

9、Group Windows

Group window 聚合根据时间或行计数间隔将行分为有限组，并为每个分组进行一次聚合函数计算。对于批处理表，窗口是按时间间隔对记录进行分组的便捷方式。

窗口是使用 window(GroupWindow w) 子句定义的，并且需要使用 as 子句来指定别名。为了按窗口对表进行分组，窗口别名的引用必须像常规分组属性一样在 groupBy(...) 子句中。以下示例展示了如何在表上定义窗口聚合。语法示例：

Table table = input
  .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
  .groupBy($("w"))  // 以窗口 w 对表进行分组
  .select($("b").sum());  // 聚合

在流环境中，如果窗口聚合除了窗口之外还根据一个或多个属性进行分组，则它们只能并行计算，例如，groupBy(...) 子句引用了一个窗口别名和至少一个附加属性。仅引用窗口别名（例如在上面的示例中）的 groupBy(...) 子句只能由单个非并行任务进行计算。以下示例展示了如何定义有附加分组属性的窗口聚合。语法示例：

Table table = input
  .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
  .groupBy($("w"), $("a"))  // 以属性 a 和窗口 w 对表进行分组
  .select($("a"), $("b").sum());  // 聚合

时间窗口的开始、结束或行时间戳等窗口属性可以作为窗口别名的属性添加到 select 子句中，如 w.start、w.end 和 w.rowtime。窗口开始和行时间戳是包含的上下窗口边界。相反，窗口结束时间戳是唯一的上窗口边界。例如，从下午 2 点开始的 30 分钟滚动窗口将 “14:00:00.000” 作为开始时间戳，“14:29:59.999” 作为行时间时间戳，“14:30:00.000” 作为结束时间戳。

语法示例：

Table table = input
  .window([GroupWindow w].as("w"))  // 定义窗口并指定别名为 w
  .groupBy($("w"), $("a"))  // 以属性 a 和窗口 w 对表进行分组
  .select($("a"), $("w").start(), $("w").end(), $("w").rowtime(), $("b").count()); // 聚合并添加窗口开始、结束和 rowtime 时间戳

Window 参数定义了如何将行映射到窗口。 Window 不是用户可以实现的接口。相反，Table API 提供了一组具有特定语义的预定义 Window 类。下面列出了支持的窗口定义。

1）、Tumble (Tumbling Windows)

滚动窗口将行分配给固定长度的非重叠连续窗口。例如，一个 5 分钟的滚动窗口以 5 分钟的间隔对行进行分组。滚动窗口可以定义在事件时间、处理时间或行数上。滚动窗口是通过 Tumble 类定义的，具体如下：

<div style="width: 18em"> <table> <thead> <tr> <th>Method </th> <th>Description</th> </tr> </thead> <tbody> <tr> <td align="left">over</td> <td align="left">将窗口的长度定义为时间或行计数间隔。</td> </tr> <tr> <td align="left">on</td> <td align="left">要对数据进行分组（时间间隔）或排序（行计数）的时间属性。批处理查询支持任意 Long 或 Timestamp 类型的属性。流处理查询仅支持声明的事件时间或处理时间属性。</td> </tr> <tr> <td align="left">as</td> <td align="left">指定窗口的别名。别名用于在 groupBy() 子句中引用窗口，并可以在 select() 子句中选择如窗口开始、结束或行时间戳的窗口属性。</td> </tr> </tbody> </table> </div>

// Tumbling Event-time Window
.window(Tumble.over(lit(10).minutes()).on($("rowtime")).as("w"));

// Tumbling Processing-time Window (assuming a processing-time attribute "proctime")
.window(Tumble.over(lit(10).minutes()).on($("proctime")).as("w"));

// Tumbling Row-count Window (assuming a processing-time attribute "proctime")
.window(Tumble.over(rowInterval(10)).on($("proctime")).as("w"));

具体示例

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.expressions.ApiExpressionUtils.unresolvedCall;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
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.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.BuiltInFunctionDefinitions;
import org.apache.flink.types.Row;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestTableAPIOperationWithWindowDemo {
	final static List<User> userList = Arrays.asList(
			new User(1L, "alan", 18, 1698742358391L), 
			new User(2L, "alan", 19, 1698742359396L), 
			new User(3L, "alan", 25, 1698742360407L),
			new User(4L, "alanchan", 28, 1698742361409L), 
			new User(5L, "alanchan", 29, 1698742362424L)
			);
	
	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private int balance;
		private Long rowtime;
	}
	
	static void testTumbleOver() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
				
		//按属性、时间窗口分组后的互异（互不相同、去重）聚合
		Table groupByWindowResult = usersTable
			    .window(Tumble
			            .over(lit(5).minutes())
			            .on($("rowtime"))
			            .as("w")
			    )
			    .groupBy($("name"), $("w"))
			    .select($("name"), $("balance").sum().distinct().as("sum_balance"));
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(groupByWindowResult, Row.class);
		result2DS.print("result2DS:");
//		result2DS::2> (true,+I[alan, 62])
//		result2DS::16> (true,+I[alanchan, 57])
		
		//使用分组窗口结合单个或者多个分组键对表进行分组和聚合。
		Table result = usersTable
			    .window(Tumble.over(lit(5).minutes()).on($("rowtime")).as("w")) // 定义窗口
			    .groupBy($("name"), $("w")) // 按窗口和键分组
			    // 访问窗口属性并聚合
			    .select(
			        $("name"),
			        $("w").start(),
			        $("w").end(),
			        $("w").rowtime(),
			        $("balance").sum().as("sum(balance)")
			    );
				
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print("resultDS:");
//		resultDS::2> (true,+I[alan, 2023-10-31T08:50, 2023-10-31T08:55, 2023-10-31T08:54:59.999, 62])
//		resultDS::16> (true,+I[alanchan, 2023-10-31T08:50, 2023-10-31T08:55, 2023-10-31T08:54:59.999, 57])

		env.execute();
	}
		
	/**
	 * @param args
	 * @throws Exception 
	 */
	public static void main(String[] args) throws Exception {
		testTumbleOver();

	}

}

2）、Slide (Sliding Windows)

滑动窗口具有固定大小并按指定的滑动间隔滑动。如果滑动间隔小于窗口大小，则滑动窗口重叠。因此，行可能分配给多个窗口。例如，15 分钟大小和 5 分钟滑动间隔的滑动窗口将每一行分配给 3 个不同的 15 分钟大小的窗口，以 5 分钟的间隔进行一次计算。滑动窗口可以定义在事件时间、处理时间或行数上。滑动窗口是通过 Slide 类定义的，具体如下：

<div style="width: 18em"> <table> <thead> <tr> <th>Method </th> <th>Description</th> </tr> </thead> <tbody> <tr> <td align="left">over</td> <td align="left">将窗口的长度定义为时间或行计数间隔。</td> </tr> <tr> <td align="left">every</td> <td align="left">将窗口的长度定义为时间或行计数间隔。滑动间隔的类型必须与窗口长度的类型相同。</td> </tr> <tr> <td align="left">on</td> <td align="left">要对数据进行分组（时间间隔）或排序（行计数）的时间属性。批处理查询支持任意 Long 或 Timestamp 类型的属性。流处理查询仅支持声明的事件时间或处理时间属性</td> </tr> <tr> <td align="left">as</td> <td align="left">指定窗口的别名。别名用于在 groupBy() 子句中引用窗口，并可以在 select() 子句中选择如窗口开始、结束或行时间戳的窗口属性。</td> </tr> </tbody> </table> </div>

// Sliding Event-time Window
.window(Slide.over(lit(10).minutes())
            .every(lit(5).minutes())
            .on($("rowtime"))
            .as("w"));

// Sliding Processing-time window (assuming a processing-time attribute "proctime")
.window(Slide.over(lit(10).minutes())
            .every(lit(5).minutes())
            .on($("proctime"))
            .as("w"));

// Sliding Row-count window (assuming a processing-time attribute "proctime")
.window(Slide.over(rowInterval(10)).every(rowInterval(5)).on($("proctime")).as("w"));

具体示例

static void testSlidingOver() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		
		Table result1 =  usersTable
				.window(
						Slide.over(lit(10).minutes())
								.every(lit(5).minutes())
								.on($("rowtime"))
								.as("w")
							)
				.groupBy($("name"),$("w"))
				.select($("name"),$("balance").sum().as("sum(balance)"),$("w").start(),$("w").end(),$("w").rowtime())
				;
		DataStream<Tuple2<Boolean, Row>> result1DS = tenv.toRetractStream(result1, Row.class);
//		result1DS.print("result1DS:");
//		result1DS::16> (true,+I[alanchan, 57, 2023-10-31T08:45, 2023-10-31T08:55, 2023-10-31T08:54:59.999])
//		result1DS::2> (true,+I[alan, 62, 2023-10-31T08:45, 2023-10-31T08:55, 2023-10-31T08:54:59.999])
//		result1DS::16> (true,+I[alanchan, 57, 2023-10-31T08:50, 2023-10-31T09:00, 2023-10-31T08:59:59.999])
//		result1DS::2> (true,+I[alan, 62, 2023-10-31T08:50, 2023-10-31T09:00, 2023-10-31T08:59:59.999])		
		
		// Sliding Processing-time window (assuming a processing-time attribute "proctime")
		Table usersTable2 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime().as("proctime"));
		Table result2 =  usersTable2
			.window(Slide.over(lit(10).minutes())
									.every(lit(5).minutes())
									.on($("proctime"))
									.as("w")
									)
			.groupBy($("name"),$("w"))
			.select($("name"),$("balance").sum().as("sum(balance)"),$("w").start(),$("w").end(),$("w").proctime())
			;
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
//		result2DS.print("result2DS:");
//		result2DS::2> (true,+I[alan, 62, 2023-10-31T08:45, 2023-10-31T08:55, 2023-11-03T02:17:19.345Z])
//		result2DS::16> (true,+I[alanchan, 57, 2023-10-31T08:45, 2023-10-31T08:55, 2023-11-03T02:17:19.345Z])
//		result2DS::16> (true,+I[alanchan, 57, 2023-10-31T08:50, 2023-10-31T09:00, 2023-11-03T02:17:19.348Z])
//		result2DS::2> (true,+I[alan, 62, 2023-10-31T08:50, 2023-10-31T09:00, 2023-11-03T02:17:19.348Z])
		
		//Sliding Row-count window (assuming a processing-time attribute "proctime")
		Table usersTable3 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime().as("proctime"));
		Table result3 =  usersTable3
				.window(Slide.over(rowInterval(10L)).every(rowInterval(5L)).on($("proctime")).as("w"))
				.groupBy($("name"),$("w"))
				.select($("name"),$("balance").sum().as("sum(balance)"))
				;
		DataStream<Tuple2<Boolean, Row>> result3DS = tenv.toRetractStream(result3, Row.class);
		result3DS.print("result3DS:");
		//Event-time grouping windows on row intervals are currently not supported.
		
		env.execute();
	}

3）、Session (Session Windows)

<div style="width: 18em"> <table> <thead> <tr> <th>Method </th> <th>Description</th> </tr> </thead> <tbody> <tr> <td align="left">withGap</td> <td align="left">将两个窗口之间的间隙定义为时间间隔</td> </tr> <tr> <td align="left">on</td> <td align="left">要对数据进行分组（时间间隔）或排序（行计数）的时间属性。批处理查询支持任意 Long 或 Timestamp 类型的属性。流处理查询仅支持声明的事件时间或处理时间属性</td> </tr> <tr> <td align="left">as</td> <td align="left">指定窗口的别名。别名用于在 groupBy() 子句中引用窗口，并可以在 select() 子句中选择如窗口开始、结束或行时间戳的窗口属性</td> </tr> </tbody> </table> </div>

// Session Event-time Window
.window(Session.withGap(lit(10).minutes()).on($("rowtime")).as("w"));

// Session Processing-time Window (assuming a processing-time attribute "proctime")
.window(Session.withGap(lit(10).minutes()).on($("proctime")).as("w"));

具体示例

static void testSessionOver() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		
		// Session Event-time Window
		Table result1 = usersTable
				.window(Session.withGap(lit(10).minutes()).on($("rowtime")).as("w"))
				.groupBy($("name"),$("w"))
				.select($("name"),$("balance").sum().as("sum(balance)"))
				; 
		
		DataStream<Tuple2<Boolean, Row>> result1DS = tenv.toRetractStream(result1, Row.class);
		result1DS.print("result1DS:");
//		result1DS::16> (true,+I[alanchan, 57])
//		result1DS::2> (true,+I[alan, 62])
		
		// Session Processing-time Window (assuming a processing-time attribute "proctime")
		Table usersTable2 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime().as("proctime"));
		Table result2 = usersTable2
					.window(Session.withGap(lit(10).minutes()).on($("proctime")).as("w"))
					.groupBy($("name"),$("w"))
					.select($("name"),$("balance").sum().as("sum(balance)"))
		;
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
		result2DS.print("result2DS:");
//		result2DS::2> (true,+I[alan, 62])
//		result2DS::16> (true,+I[alanchan, 57])
		
		env.execute();
	}

10、Over Windows

Over window 聚合聚合来自在标准的 SQL（OVER 子句），可以在 SELECT 查询子句中定义。与在“GROUP BY”子句中指定的 group window 不同， over window 不会折叠行。相反，over window 聚合为每个输入行在其相邻行的范围内计算聚合。

Over windows 使用 window(w: OverWindow*) 子句（在 Python API 中使用 over_window(*OverWindow)）定义，并通过 select() 方法中的别名引用。以下示例显示如何在表上定义 over window 聚合。

Table table = input
  .window([OverWindow w].as("w"))           // define over window with alias w
  .select($("a"), $("b").sum().over($("w")), $("c").min().over($("w"))); // aggregate over the over window w

OverWindow 定义了计算聚合的行范围。OverWindow 不是用户可以实现的接口。相反，Table API 提供了Over 类来配置 over window 的属性。可以在事件时间或处理时间以及指定为时间间隔或行计数的范围内定义 over window 。可以通过 Over 类（和其他类）上的方法来定义 over window，具体如下： <div style="width: 18em"> <table> <thead> <tr> <th>Method </th> <th>Description</th> <th>是否必选</th> </tr> </thead> <tbody> <tr> <td align="left">Partition By</td> <td align="left">在一个或多个属性上定义输入的分区。每个分区单独排序，聚合函数分别应用于每个分区。<br>在流环境中，如果窗口包含 partition by 子句，则只能并行计算 over window 聚合。如果没有 partitionBy(…)，数据流将由单个非并行任务处理。</td> <td align="left">可选的</td> </tr> <tr> <td align="left">Order By</td> <td align="left">定义每个分区内行的顺序，从而定义聚合函数应用于行的顺序。<br>对于流处理查询，必须声明事件时间或处理时间属性。目前，仅支持单个排序属性。</td> <td align="left">必须的</td> </tr> <tr> <td align="left">Preceding</td> <td align="left">定义了包含在窗口中并位于当前行之前的行的间隔。间隔可以是时间或行计数间隔。<br>有界 over window 用间隔的大小指定，例如，时间间隔为10分钟或行计数间隔为10行。<br>无界 over window 通过常量来指定，例如，用UNBOUNDED_RANGE指定时间间隔或用 UNBOUNDED_ROW 指定行计数间隔。无界 over windows 从分区的第一行开始。<br>如果省略前面的子句，则使用 UNBOUNDED_RANGE 和 CURRENT_RANGE 作为窗口前后的默认值。</td> <td align="left">可选的</td> </tr> <tr> <td align="left">Following</td> <td align="left">定义包含在窗口中并在当前行之后的行的窗口间隔。间隔必须以与前一个间隔（时间或行计数）相同的单位指定。<br>截至版本1.17，不支持在当前行之后有行的 over window。<br>可以指定两个常量之一：<br>CURRENT_ROW 将窗口的上限设置为当前行<br>CURRENT_RANGE 将窗口的上限设置为当前行的排序键，例如，与当前行具有相同排序键的所有行都包含在窗口中<br>如果省略后面的子句，则时间间隔窗口的上限定义为 CURRENT_RANGE，行计数间隔窗口的上限定义为CURRENT_ROW></td> <td align="left">可选的</td> </tr> <tr> <td align="left">As</td> <td align="left">为 over window 指定别名。别名用于在之后的 select() 子句中引用该 over window。<br>截至版本1.17，同一个 select() 调用中的所有聚合函数必须在同一个 over window 上计算。</td> <td align="left">必须的</td> </tr> </tbody> </table> </div>

1)、Unbounded Over Windows

示例代码

// 无界的事件时间 over window（假定有一个叫“rowtime”的事件时间属性）
.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(UNBOUNDED_RANGE).as("w"));

// 无界的处理时间 over window（假定有一个叫“proctime”的处理时间属性）
.window(Over.partitionBy($("a")).orderBy("proctime").preceding(UNBOUNDED_RANGE).as("w"));

// 无界的事件时间行数 over window（假定有一个叫“rowtime”的事件时间属性）
.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(UNBOUNDED_ROW).as("w"));
 
// 无界的处理时间行数 over window（假定有一个叫“proctime”的处理时间属性）
.window(Over.partitionBy($("a")).orderBy($("proctime")).preceding(UNBOUNDED_ROW).as("w"));

具体示例

	static void testOverWithUnbounded() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		
		//	所有的聚合必须定义在同一个窗口上，比如同一个分区、排序和范围内。目前只支持 PRECEDING 到当前行范围（无界或有界）的窗口。
		// 尚不支持 FOLLOWING 范围的窗口。ORDER BY 操作必须指定一个单一的时间属性。
		Table result = usersTable
			    // 定义窗口
			    .window(
			        Over
			          .partitionBy($("name"))
			          .orderBy($("rowtime"))
			          .preceding(unresolvedCall(BuiltInFunctionDefinitions.UNBOUNDED_RANGE))
			          .following(unresolvedCall(BuiltInFunctionDefinitions.CURRENT_RANGE))
			          .as("w"))
			    // 滑动聚合
			    .select(
			        $("id"),
			        $("balance").avg().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		2> (true,+I[1, 18, 18, 18])
//		16> (true,+I[4, 28, 28, 28])
//		2> (true,+I[2, 18, 19, 18])
//		16> (true,+I[5, 28, 29, 28])
//		2> (true,+I[3, 20, 25, 18])
		
		//over window 上的互异（互不相同、去重）聚合
		Table result2 = usersTable
			    .window(Over
			        .partitionBy($("name"))
			        .orderBy($("rowtime"))
			        .preceding(unresolvedCall(BuiltInFunctionDefinitions.UNBOUNDED_RANGE))
			        .as("w"))
			    .select(
			        $("name"), $("balance").avg().distinct().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		DataStream<Tuple2<Boolean, Row>> result3DS = tenv.toRetractStream(result2, Row.class);
		result3DS.print();
//		16> (true,+I[alanchan, 28, 28, 28])
//		2> (true,+I[alan, 18, 18, 18])
//		2> (true,+I[alan, 18, 19, 18])
//		16> (true,+I[alanchan, 28, 29, 28])
//		2> (true,+I[alan, 20, 25, 18])
		
		env.execute();
	}

2)、Bounded Over Windows

示例代码

// 有界的事件时间 over window（假定有一个叫“rowtime”的事件时间属性）
.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(lit(1).minutes()).as("w"));

// 有界的处理时间 over window（假定有一个叫“proctime”的处理时间属性）
.window(Over.partitionBy($("a")).orderBy($("proctime")).preceding(lit(1).minutes()).as("w"));

// 有界的事件时间行数 over window（假定有一个叫“rowtime”的事件时间属性）
.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(rowInterval(10)).as("w"));
 
// 有界的处理时间行数 over window（假定有一个叫“proctime”的处理时间属性）
.window(Over.partitionBy($("a")).orderBy($("proctime")).preceding(rowInterval(10)).as("w"));

具体示例

static void testOverWithBounded() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		// 有界的事件时间 over window（假定有一个叫“rowtime”的事件时间属性）
		//	.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(lit(1).minutes()).as("w"));
		Table result = usersTable
			    // 定义窗口
			    .window(
			        Over
			          .partitionBy($("name"))
			          .orderBy($("rowtime"))
			          .preceding(lit(1).minutes())
			          .as("w"))
			    // 滑动聚合
			    .select(
			        $("id"),
			        $("balance").avg().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
//		resultDS.print();
//		2> (true,+I[1, 18, 18, 18])
//		16> (true,+I[4, 28, 28, 28])
//		2> (true,+I[2, 18, 19, 18])
//		16> (true,+I[5, 28, 29, 28])
//		2> (true,+I[3, 20, 25, 18])
		
		// 有界的处理时间 over window（假定有一个叫“proctime”的处理时间属性）
//		.window(Over.partitionBy($("a")).orderBy($("proctime")).preceding(lit(1).minutes()).as("w"));
		Table usersTable2 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		Table result2 = usersTable2
				.window(Over.partitionBy($("name")).orderBy($("rowtime")).preceding(lit(1).minutes()).as("w"))
			    .select(
			        $("id"),
			        $("balance").avg().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		
		DataStream<Tuple2<Boolean, Row>> result2DS = tenv.toRetractStream(result2, Row.class);
//		result2DS.print();
//		16> (true,+I[4, 28, 28, 28])
//		2> (true,+I[1, 18, 18, 18])
//		2> (true,+I[2, 18, 19, 18])
//		16> (true,+I[5, 28, 29, 28])
//		2> (true,+I[3, 20, 25, 18])
		
		// 有界的事件时间行数 over window（假定有一个叫“rowtime”的事件时间属性）
		//.window(Over.partitionBy($("a")).orderBy($("rowtime")).preceding(rowInterval(10)).as("w"));
		Table usersTable3 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		Table result3 = usersTable3
				.window(Over.partitionBy($("name")).orderBy($("rowtime")).preceding(rowInterval(10L)).as("w"))
//				.window(Over.partitionBy($("name")).orderBy($("rowtime")).preceding(lit(1).minutes()).as("w"))
			    .select(
			        $("id"),
			        $("balance").avg().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		
		DataStream<Tuple2<Boolean, Row>> result3DS = tenv.toRetractStream(result3, Row.class);
		result3DS.print("result3DS:");
//		result3DS::16> (true,+I[4, 28, 28, 28])
//		result3DS::2> (true,+I[1, 18, 18, 18])
//		result3DS::16> (true,+I[5, 28, 29, 28])
//		result3DS::2> (true,+I[2, 18, 19, 18])
//		result3DS::2> (true,+I[3, 20, 25, 18])
		
		// 有界的处理时间行数 over window（假定有一个叫“proctime”的处理时间属性）
		//	.window(Over.partitionBy($("a")).orderBy($("proctime")).preceding(rowInterval(10)).as("w"));
		Table usersTable4 = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		Table result4 = usersTable4
				.window(Over.partitionBy($("name")).orderBy($("rowtime")).preceding(rowInterval(10L)).as("w"))
//				.window(Over.partitionBy($("name")).orderBy($("rowtime")).preceding(lit(1).minutes()).as("w"))
			    .select(
			        $("id"),
			        $("balance").avg().over($("w")),
			        $("balance").max().over($("w")),
			        $("balance").min().over($("w"))
			    );
		
		DataStream<Tuple2<Boolean, Row>> result4DS = tenv.toRetractStream(result4, Row.class);
		result4DS.print("result4DS:");
//		result4DS::16> (true,+I[4, 28, 28, 28])
//		result4DS::16> (true,+I[5, 28, 29, 28])
//		result4DS::2> (true,+I[1, 18, 18, 18])
//		result4DS::2> (true,+I[2, 18, 19, 18])
//		result4DS::2> (true,+I[3, 20, 25, 18])
		
		env.execute();
	}

11、Row-based Operations

由于文章太长，将公共代码部分放在下面表示，具体示例在每个子项中以方法的形式展示。将每个子项的方法拷贝至下面的公共代码中即可运行。

import static org.apache.flink.table.api.Expressions.$;
import static org.apache.flink.table.api.Expressions.call;
import static org.apache.flink.table.api.Expressions.lit;
import static org.apache.flink.table.api.Expressions.row;

import java.time.Duration;
import java.util.Arrays;
import java.util.List;

import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.typeutils.RowTypeInfo;
import org.apache.flink.streaming.api.datastream.DataStream;
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.Tumble;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.table.functions.ScalarFunction;
import org.apache.flink.table.functions.TableAggregateFunction;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import lombok.AllArgsConstructor;
import lombok.Data;
import lombok.NoArgsConstructor;

/**
 * @author alanchan
 *
 */
public class TestTableAPIOperationWithRowbasedDemo {
	@Data
	@NoArgsConstructor
	@AllArgsConstructor
	public static class User {
		private long id;
		private String name;
		private int balance;
		private Long rowtime;
	}

	final static List<User> userList = Arrays.asList(
			new User(1L, "alan", 18, 1698742358391L), 
			new User(2L, "alan", 19, 1698742359396L), 
			new User(3L, "alan", 25, 1698742360407L),
			new User(4L, "alanchan", 28, 1698742361409L), 
			new User(5L, "alanchan", 29, 1698742362424L)
			);

	public static class MyMapFunction extends ScalarFunction {

		public Row eval（String a) {
			return Row.of(a, "pre-" + a);
		}

		@Override
		public TypeInformation<?> getResultType(Class<?>[] signature) {
			return Types.ROW(Types.STRING, Types.STRING);
		}
	}

	/**
	 * @param args
	 * @throws Exception
	 */
	public static void main(String[] args) throws Exception {
//		testMap();
//		testFlatMap();
//		testAggregate();
//		testGroupWindowAggregate();
		testFlatAggregate();
	}

}

基于行生成多列输出的操作。

1）、Map

使用用户定义的标量函数或内置标量函数执行 map 操作。如果输出类型是复合类型，则输出将被展平。

public static class MyMapFunction extends ScalarFunction {

		public Row eval（String a) {
			return Row.of(a, "pre-" + a);
		}

		@Override
		public TypeInformation<?> getResultType(Class<?>[] signature) {
			return Types.ROW(Types.STRING, Types.STRING);
		}
	}

	/**
	 * 使用用户定义的标量函数或内置标量函数执行 map 操作。如果输出类型是复合类型，则输出将被展平。
	 * @throws Exception
	 */
	static void testMap() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		ScalarFunction func = new MyMapFunction();
		tenv.registerFunction("func", func);

//		DataStream<String> users = env.fromCollection(Arrays.asList("alan", "alanchan", "alanchanchn"));
//		Table usersTable = tenv.fromDataStream(users, $("name"));

		DataStream<User> users = env.fromCollection(userList);
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"), $("rowtime"));

		Table result = usersTable.map(call("func", $("name")));

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		2> (true,+I[alan, pre-alan])
//		4> (true,+I[alan, pre-alan])
//		6> (true,+I[alanchan, pre-alanchan])
//		5> (true,+I[alanchan, pre-alanchan])
//		3> (true,+I[alan, pre-alan])

		env.execute();
	}

2）、FlatMap

使用表函数执行 flatMap 操作。

public static class MyFlatMapFunction extends TableFunction<Row> {

		public void eval（String str) {
			if (str.contains("#")) {
				String[] array = str.split("#");
				for (int i = 0; i < array.length; ++i) {
					collect(Row.of(array[i], array[i].length()));
				}
			}
		}

		@Override
		public TypeInformation<Row> getResultType() {
			return Types.ROW(Types.STRING, Types.INT);
		}
	}

	/**
	 * 使用表函数执行 flatMap 操作。
	 * 
	 * @author alanchan
	 *
	 */
	static void testFlatMap() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		TableFunction func = new MyFlatMapFunction();
		tenv.registerFunction("func", func);

//		DataStream<User> users = env.fromCollection(userList);
//		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"), $("rowtime"));
		DataStream<String> users = env.fromCollection(Arrays.asList("alan#alanchan#alanchanchn", "alan_chan_chn#", "alan-chan-chn"));
		Table usersTable = tenv.fromDataStream(users, $("name"));

		Table result = usersTable.flatMap(call("func", $("name")));

		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		13> (true,+I[alan_chan_chn, 13])
//		10> (true,+I[alan, 4])
//		12> (true,+I[alanchanchn, 11])
//		11> (true,+I[alanchan, 8])

		env.execute();
	}

3）、Aggregate

使用聚合函数来执行聚合操作。你必须使用 select 子句关闭 aggregate，并且 select 子句不支持聚合函数。如果输出类型是复合类型，则聚合的输出将被展平

public static class MyMinMaxAcc {
		public int min = 0;
		public int max = 0;
	}

	public static class MyMinMax extends AggregateFunction<Row, MyMinMaxAcc> {
		public void accumulate(MyMinMaxAcc acc, int value) {
			if (value < acc.min) {
				acc.min = value;
			}
			if (value > acc.max) {
				acc.max = value;
			}
		}

		@Override
		public MyMinMaxAcc createAccumulator() {
			return new MyMinMaxAcc();
		}

		public void resetAccumulator(MyMinMaxAcc acc) {
			acc.min = 0;
			acc.max = 0;
		}

		@Override
		public Row getValue(MyMinMaxAcc acc) {
			return Row.of(acc.min, acc.max);
		}

		@Override
		public TypeInformation<Row> getResultType() {
			return new RowTypeInfo(Types.INT, Types.INT);
		}
	}

	/**
	 * 使用聚合函数来执行聚合操作。你必须使用 select 子句关闭 aggregate，并且 select 子句不支持聚合函数。如果输出类型是复合类型，则聚合的输出将被展平。
	 * 
	 * @throws Exception
	 */
	static void testAggregate() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);

		AggregateFunction myAggFunc = new MyMinMax();
		tenv.registerFunction("myAggFunc", myAggFunc);

		Table ordersTable = tenv.fromValues(
				DataTypes.ROW(
//						DataTypes.FIELD("key", DataTypes.BIGINT()),
						DataTypes.FIELD("name", DataTypes.STRING()),
						DataTypes.FIELD("balance", DataTypes.INT())
						),
				Arrays.asList(
						row("alan", 16987423), 
						row("alan", 16396), 
						row("alanchan", 1690407),
						row("alanchanchn", 16409), 
						row("alanchan", 162424),
						row("alan", 164)
						));
				
		Table usersTable = ordersTable.select($("name"),$("balance"));
		
//		Table usersTable = tenv.fromDataStream(users, $("key"),$("name"),$("age"));
		
		Table result = usersTable
				.groupBy($("name")).
				aggregate(call("myAggFunc", $("balance")))
				.select($("name"), $("f0"),$("f1"));
		
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		2> (true,+I[alan, 0, 16987423])
//		16> (true,+I[alanchan, 0, 1690407])
//		16> (true,+I[alanchanchn, 0, 16409])
		
		env.execute();
	}

4）、Group Window Aggregate

在 group window 和可能的一个或多个分组键上对表进行分组和聚合。你必须使用 select 子句关闭 aggregate。并且 select 子句不支持“*“或聚合函数。

/**
	 * 在 group window 和可能的一个或多个分组键上对表进行分组和聚合。你必须使用 select 子句关闭 aggregate。并且 select 子句不支持“*“或聚合函数。
	 * 
	 * @throws Exception
	 */
	static void testGroupWindowAggregate() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		
		AggregateFunction myAggFunc = new MyMinMax();
		tenv.registerFunction("myAggFunc", myAggFunc);
		
		List<User> userList = Arrays.asList(
				new User(1L, "alan", 18, 1698742358391L), 
				new User(2L, "alan", 19, 1698742359396L), 
				new User(3L, "alan", 25, 1698742360407L),
				new User(4L, "alanchan", 28, 1698742361409L), 
				new User(5L, "alanchan", 29, 1698742362424L),
				new User(5L, "alanchan", 29, 1698742362424L)
				);
		
		DataStream<User> users = env.fromCollection(userList)
				.assignTimestampsAndWatermarks(
						WatermarkStrategy
						.<User>forBoundedOutOfOrderness(Duration.ofSeconds(1))
						.withTimestampAssigner((user, recordTimestamp) -> user.getRowtime())
						);
		
		Table usersTable = tenv.fromDataStream(users, $("id"), $("name"), $("balance"),$("rowtime").rowtime());
		
		Table result = usersTable
			    .window(Tumble.over(lit(5).minutes())
			                  .on($("rowtime"))
			                  .as("w")) // 定义窗口
			    .groupBy($("name"), $("w")) // 以键和窗口分组
			    .aggregate(call("myAggFunc", $("balance")))
			    .select($("name"), $("f0"), $("f1"), $("w").start(), $("w").end()); // 访问窗口属性与聚合结果
		
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		2> (true,+I[alan, 0, 25, 2023-10-31T08:50, 2023-10-31T08:55])
//		16> (true,+I[alanchan, 0, 29, 2023-10-31T08:50, 2023-10-31T08:55])
		
		env.execute();
	}

5）、FlatAggregate

和 GroupBy Aggregation 类似。使用运行中的表之后的聚合算子对分组键上的行进行分组，以按组聚合行。和 AggregateFunction 的不同之处在于，TableAggregateFunction 的每个分组可能返回0或多条记录。你必须使用 select 子句关闭 flatAggregate。并且 select 子句不支持聚合函数。

除了使用 emitValue 输出结果，你还可以使用 emitUpdateWithRetract 方法。和 emitValue 不同的是，emitUpdateWithRetract 用于下发已更新的值。此方法在retract 模式下增量输出数据，例如，一旦有更新，我们必须在发送新的更新记录之前收回旧记录。如果在表聚合函数中定义了这两个方法，则将优先使用 emitUpdateWithRetract 方法而不是 emitValue 方法，这是因为该方法可以增量输出值，因此被视为比 emitValue 方法更有效。

/**
	 * Top2 Accumulator。
	 */
	public static class Top2Accum {
	    public Integer first;
	    public Integer second;
	}

	/**
	 * 用户定义的聚合函数 top2。
	 */
	public static  class Top2 extends TableAggregateFunction<Tuple2<Integer, Integer>, Top2Accum> {

	    @Override
	    public Top2Accum createAccumulator() {
	        Top2Accum acc = new Top2Accum();
	        acc.first = Integer.MIN_VALUE;
	        acc.second = Integer.MIN_VALUE;
	        return acc;
	    }


	    public void accumulate(Top2Accum acc, Integer v) {
	        if (v > acc.first) {
	            acc.second = acc.first;
	            acc.first = v;
	        } else if (v > acc.second) {
	            acc.second = v;
	        }
	    }

	    public void merge(Top2Accum acc, java.lang.Iterable<Top2Accum> iterable) {
	        for (Top2Accum otherAcc : iterable) {
	            accumulate(acc, otherAcc.first);
	            accumulate(acc, otherAcc.second);
	        }
	    }

	    public void emitValue(Top2Accum acc, Collector<Tuple2<Integer, Integer>> out) {
	        // 下发 value 与 rank
	        if (acc.first != Integer.MIN_VALUE) {
	            out.collect(Tuple2.of(acc.first, 1));
	        }
	        if (acc.second != Integer.MIN_VALUE) {
	            out.collect(Tuple2.of(acc.second, 2));
	        }
	    }
	}
	
	/**
	 * 和 GroupBy Aggregation 类似。使用运行中的表之后的聚合算子对分组键上的行进行分组，以按组聚合行。
	 * 和 AggregateFunction 的不同之处在于，TableAggregateFunction 的每个分组可能返回0或多条记录。
	 * 必须使用 select 子句关闭 flatAggregate。并且 select 子句不支持聚合函数。
	 * 
	 * @throws Exception
	 */
	static void testFlatAggregate() throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
		env.setParallelism(1);
		tenv.registerFunction("top2", new Top2());
		
		Table ordersTable = tenv.fromValues(
				DataTypes.ROW(
//						DataTypes.FIELD("key", DataTypes.BIGINT()),
						DataTypes.FIELD("name", DataTypes.STRING()),
						DataTypes.FIELD("balance", DataTypes.INT())
						),
				Arrays.asList(
						row("alan", 16987423), 
						row("alan", 16396), 
						row("alanchan", 1690407),
						row("alanchanchn", 16409), 
						row("alanchan", 162424),
						row("alan", 164)
						));
		
//		Table orders = tenv.from("Orders");
		
		Table result = ordersTable
			    .groupBy($("name"))
			    .flatAggregate(call("top2", $("balance")))
			    .select($("name"), $("f0").as("balance"), $("f1").as("rank"));
			    
		DataStream<Tuple2<Boolean, Row>> resultDS = tenv.toRetractStream(result, Row.class);
		resultDS.print();
//		(true,+I[alan, 16987423, 1])
//		(false,-D[alan, 16987423, 1])
//		(true,+I[alan, 16987423, 1])
//		(true,+I[alan, 16396, 2])
//		(true,+I[alanchan, 1690407, 1])
//		(true,+I[alanchanchn, 16409, 1])
//		(false,-D[alanchan, 1690407, 1])
//		(true,+I[alanchan, 1690407, 1])
//		(true,+I[alanchan, 162424, 2])
//		(false,-D[alan, 16987423, 1])
//		(false,-D[alan, 16396, 2])
//		(true,+I[alan, 16987423, 1])
//		(true,+I[alan, 16396, 2])
		
		env.execute();	    
	}

二、数据类型

数据类型，请参考文章：14、Flink 的table api与sql之数据类型: 内置数据类型以及它们的属性

行中的字段可以是一般类型和(嵌套)复合类型(比如 POJO、元组、行、 Scala 案例类)。

任意嵌套的复合类型的字段都可以通过值访问函数来访问。

用户自定义函数可以将泛型当作黑匣子一样传输和处理。

关于函数请参考文章： 19、Flink 的Table API 和 SQL 中的内置函数（1） 19、Flink 的Table API 和 SQL 中的自定义函数（2）

以上，通过示例介绍了如何使用table api进行表的联接、排序、insert、group window、over window、以及基于行的操作，每个示例都是经过验证的、可运行的，并将运行结果展示在输出后面。

标签：rowtime,17,id,API,Table,DataTypes,true,tenv
From： https://blog.51cto.com/alanchan2win/8173483