Flink table API 操作

我会带着你远行 2021-11-22 03:02 338阅读 0赞

Table API是用于流和批处理的统一关系API。表API查询可以在批量或流式输入上运行而无需修改。Table API是SQL语言的超级集合，专门用于Apache Flink。Table API是Scala和Java的语言集成API。Table API查询不是像SQL中常见的那样将查询指定为String值，而是在Java或Scala中以嵌入语言的样式定义，具有IDE支持，如自动完成和语法验证。

Table API可用于Scala和Java。Scala Table API利用Scala表达式，Java Table API基于字符串，这些字符串被解析并转换为等效表达式。

以下示例显示了Scala和Java Table API之间的差异。表程序在批处理环境中执行。它 Orders按字段扫描表，并按组计算结果行。表程序的结果转换为DataSet 类型 Row 并打印。

通过导入启用Java Table API org.apache.flink.table.api.java.\* 。以下示例显示如何构造Java Table API程序以及如何将表达式指定为字符串。

ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();	
    BatchTableEnvironment tEnv = BatchTableEnvironment.create(env);	
    	
    Table orders = tEnv.scan("Orders"); // schema (a, b, c, rowtime)	
    	
    Table counts = orders	
            .groupBy("a")	
            .select("a, b.count as cnt");	
    	
    DataSet<Row> result = tEnv.toDataSet(counts, Row.class);	
    result.print();

下一个示例显示了一个更复杂的Table API程序。程序再次扫描 Orders 表格。它过滤空值，规范化a String类型的字段，并计算每小时和产品a 的平均计费金额 b。

Table orders = tEnv.scan("Orders");	
    	
    Table result = orders	
            .filter("a.isNotNull && b.isNotNull && c.isNotNull")	
            .select("a.lowerCase() as a, b, rowtime")	
            .window(Tumble.over("1.hour").on("rowtime").as("hourlyWindow"))	
            .groupBy("hourlyWindow, a")	
            .select("a, hourlyWindow.end as hour, b.avg as avgBillingAmount");

由于Table API是批处理和流数据的统一API，因此两个示例程序都可以在批处理和流输入上执行，而无需对表程序本身进行任何修改。

![640?wx\_fmt=jpeg][640_wx_fmt_jpeg]

[640_wx_fmt_jpeg]: /images/20211122/f9d9412ea8bf4449b3b7d16607059a22.png