ApacheFlinkTableAPI的Catalog

“ Apache Flink的Table API提供了对数据注册为Table的方式， 实现把数据通过SQL的方式进行计算。Table API与SQL API实现了Apache Flink的批流统一的实现方式。Table API与SQL API的核心概念就是TableEnviroment。TableEnviroment对象提供方法注册数据源与数据表信息。那么数据源与数据表的信息则存储在CataLog中。所以，CataLog是TableEnviroment的重要组成部分。”

1. 内部Table注册
   通过TableEnviroment的Register接口对数据进行注册。TableEnviroment的registerTable包含两个参数("tableName",table)。tableName就是注册在CataLog中的表名。第二个参数table则是对应的Table对象。Table则是由TableEnviroment生成得来或者是通过DataSet、DataStream转换的来。

...
tableEnv.registerDataSet("USER",dataset,'name,'age)

val result = tableEnv.sqlQuery("SELECT name,sum(age) FROM `USER` GROUP BY name")
//使用Table对象注册Table
tableEnv.registerTable("TABLE_RES",table)
//输出注册的Table中的内容
tableEnv.sqlQuery("SELECT * FROM `TABLE_RES`").toDataSet[Row].print()
...


1. TableSource注册
   通过TableSource对数据外部的数据源注册为Table数据结构。例如常见的有csv，Text，Parquet等文件格式。例如下代码，通过外部csv数据源注册为Table数据。然后可以通过SQL API对数据进行检索。

val env = StreamExecutionEnvironment.getExecutionEnvironment
val tableEnv = StreamTableEnvironment.create(env)
//使用CSV的方式进行注册表结构.
// 参数(path:数据的路径地址,fieldNames:字段名称,fieldTypes:字段类型,fieldDelim:csv分隔符,rowDelim:行分割方式)
val csvTableSource:CsvTableSource =new CsvTableSource("../datas.csv",Array("exitcode","count"),Array(Types.STRING,Types.INT),",","\n")

tableEnv.registerTableSource("csv",csvTableSource)



1. TableSink注册
   当数据处理、计算完成后就需要写入到外部的数据中。外部数据包括文本(CSV、Apache[Parquet、ORC、Avro])、数据库、KV库、消息队列等。Apche Flink通过Table Sink用于支持常见的数据存储格式与存储系统。

val csvPath = "D:/flink.csv"
val fieldNames = Array[String]("user","age")
val fieldTypes = Array[TypeInformation[_]](Types.STRING,Types.INT)

val csvSink:CsvTableSink = new CsvTableSink(csvPath,",")

tableEnv.registerTableSink("csv",fieldNames,fieldTypes,csvSink)

tableEnv.sqlQuery("SELECT * FROM `USER` ").insertInto("csv")

val memoryCataLog:ExternalCatalog = new InMemoryExternalCatalog("UserCataLog")

tableEnv.registerExternalCatalog("user",memoryCataLog)



⬇⬇⬇  你好,我是CainGao。一线大数据开发者，关注我一起交流场景实现  ⬇⬇⬇