ApacheFlinkTableAPI的Catalog

一文了解ApacheFlinkCa

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

Apache Flink在获取TableEnviroment对象后，可以通过Register实现对数据源与数据表进行注册。注册完成后数据库与数据表的原信息则存储在CataLog中。CataLog中保存了所有的表结构信息、数据目录信息等。

内部CataLog注册

内部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对象注册TabletableEnv.registerTable("TABLE_RES",table)//输出注册的Table中的内容tableEnv.sqlQuery("SELECT * FROM `TABLE_RES`").toDataSet[Row].print()...

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

val env = StreamExecutionEnvironment.getExecutionEnvironmentval 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)以上使用的是StreamExecutionEnvironment进行的处理，当然也可以使用Batch的方式对数据进行注册可以自己进行尝试。

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")

外部CataLog注册

Apache Flink除了实现内部的CataLog作为所有Table的元数据存储介质之外还可以把CataLog放到其他的存储介质中。外部的CataLog可以自定义实现，然后在TableEnvironment中注册实现。Apache Flink官方提供了InMemoryCataLog的实现，开发者可以参考来实现其他的存储介质的CataLog。

val memoryCataLog:ExternalCatalog = new InMemoryExternalCatalog("UserCataLog")tableEnv.registerExternalCatalog("user",memoryCataLog)以上为Apache Flink的CataLog的实现。阿里云Blink中对CataLog进行了重构，并且增加了两种cataLog。一种是基于内存的InMemoryCataLog，另外一种是能够桥接Hive metaStore的Hive CataLog。使Flink能够直接读取Hive的metaStore信息，打通Hive的数据连接。

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