Flink Table API 的开发步骤

标签：TableEnvironment Flink API table Table tEnv

1.创建 TableEnvironment

Flink的Table API/SQL的执行入口

功能如下：

1.1 注册Catalog【数据系统实例】

1.2 在Catalog中注册库和表

1.3 加载插件模块

1.4 执行SQL的查询

1.5 注册UDF

1.6 实现DataStream和Table互转

方式一：创建TableEnvironment实例代码，：

EnvironmentSettings settings = EnvironmentSettings
.newInstance() //创建对象
// .inBatchMode() // 批模式环境
// .inStreamingMode() //流模式环境【默认】
.build();

TableEnvironment tableEnvironment = TableEnvironment.create(settings);

方式二：DataStream和Table 的API/SQL 混用，可以这种创建

// 1.获取Stream执行环境
 StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
// 2.创建表执行环境
StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

2.修改Flink的默认配置，可以参考官网：https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/config/

Configuration configuration = tEnv.getConfig().getConfiguration();
configuration.setString("table.exec.mini-batch.enabled", "true");
configuration.setString("table.exec.mini-batch.allow-latency", "5 s");
configuration.setString("table.exec.mini-batch.size", "5000");

3.创建表和注册表

// 临时表
tEnv.createTemporaryView(); 
// 常规表
tEnv.createTable();

tEnv.createTemporaryTable();

常规表一般是创建source和sink的时候使用，可以跨session使用

临时表一般是中间表，中间过程表，只能在同一个session使用

4.设置表的标识

catalog【数据库实例】，database【数据库】，如果不指定默认 default_catalog ,default_database

tEnv.useCatalog("");
tEnv.useDatabase("");

5.执行查询

Table的API方式

Table resultTable = table.where($("username").isEqual("zhangsan"))
.select($("username"), $("url"),$("ctime"));

SQL的方式

Table table1 = tenv.sqlQuery("select * from order_sink");

6.将结果打印/输出

clickLogDS.print();

7.执行

senv.execute("FlinkSQLOracleDemo");

 

标签：TableEnvironment,Flink,API,table,Table,tEnv
From： https://www.cnblogs.com/glblog/p/16783089.html