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