第四章 大数据交互式OLAP多维分析开发实践-4.5分布式计算框架Flink关系型API.pptxVIP

第四章

大数据交互式

OLAP多维分析开发实践

大数据交互式分析技术栈分布式数据仓库HiveHiveSQL操作实践分布式计算框架SparkSQL分布式计算框架Flink关系型API大数据交互式OLAP多维分析实践

4.5.1Flink的关系型API概述

及实现原理

Flink的关系型API概述及实现原理Flink提供两种顶层的关系型API，分别为TableAPI和SQL。Flink通过TableAPI和SQL实现批流统一。Flink提供的TableAPI是用于Python、Scala和Java的语言集成查询API，它允许以非常直观的方式组合关系运算符（例如select、where和join）进行查询。Flink对SQL的支持基于实现了SQL标准的Calcite。无论数据输入是有界的（批处理）还是无界的（流处理），在任意一个接口中指定的查询都具有相同的语义并具有相同的结果。TableAPI和SQL接口与Flink的DataStreamAPI无缝集成，可以轻松地在基于构建的全部API和库间进行切换。例如，可以使用MATCH_RECOGNIZE子句在表中检测模式，然后使用DataStreamAPI基于检测到的模式构建alerting。

TableEnvironment对象TableEnvironment表环境是TableAPI和SQL集成的核心概念，使用TableAPI或SQL创建Flink应用程序，需要在环境中创建TableEnvironment对象。TableEnvironment对象提供注册内部表、注册外部目录、执行FlinkSQL、注册自定义函数以及将DataStream或DataSet转换为Table等功能。一张表始终与某个特定的TableEnvironment对象绑定，一个在查询中组合的表只能是具有同一个TableEnvironment的表。批式应用创建ExecutionEnvironment，通过BatchTableEnvironment.create()创建BatchTableEnvironment对象，代码如下。流应用创建ExecutionEnvironment，通过StreamTableEnvironment.create()创建StreamTableEnvironment对象，代码如下。ExecutionEnvironmentenv=ExecutionEnvironment.getExecutionEnvironment();//使用ExecutionEnvironment创建BatchTableEnvironmentBatchTableEnvironmenttableEnvironment=BatchTableEnvironment.create(env);StreamExecutionEnvironmentenv=StreamExecutionEnvironment.getExecutionEnvironment();//使用StreamExecutionEnvironment创建StreamTableEnvironmentStreamTableEnvironmenttableEnvironment=StreamTableEnvironment.create(env);

注册表每张表都有一个catalog目录，每次注册一张表，都会在catalog目录中注册相应的表信息。表主要有两种类型：输入表和输出表。可以在Table或者SQL查询中引用输入表并提供输入数据，然后在输出表中将查询结果发送给外部系统保存。输入表可以从各种来源注册。如将表注册到TableEnvironment；对外连接数据源，通过数据源表的注册获取外部数据源，例如MySQL、Oracle、Kafka和一些CSV文件中的数据；从外部系统获取数据并注册一张数据保存表，将数据保存到某个位置，如MySQL、Oracle和某些文件。利用pojoType方法将外部CSV文件的数据映射为java类型，同时转换为Flink的DataSource，然后调用fromDataSet方法将数据集转换成Table。Stringpath=src/main/java/beans/sale.csv//基于位置指定对应字段名称DataSourceSalessalesDataSource=env.readCsvFile(path).ignoreFirstLine().pojoType(Sales.class,transactionId,customerId,itemId,amountPaid);//将DataSource转换成TableTabletable=tableEnvironment.fromDataS