《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门

这篇具有很好参考价值的文章主要介绍了《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

第四章中介绍了 DataStream API 以及 DataSet API 的入门案例,本章开始介绍 Table API 以及基于此的高层应用 Flink SQL 的基础。

5.1 Flink Table & SQL 基础知识

Flink 提供了两个关系API——Table API 和 SQL——用于统一的流和批处理。Table API 是一种针对Java、Scala和Python的语言集成查询API,它允许以非常直观的方式组合来自关系运算符(如选择、筛选和联接)的查询。Flink的SQL支持基于Apache Calcite,后者实现了SQL标准。无论输入是连续的(流式)还是有界的(批处理),在任一接口中指定的查询都具有相同的语义并指定相同的结果。

Table API和SQL接口与彼此以及Flink的DataStream API无缝集成。您可以轻松地在所有API和基于它们构建的库之间切换。

SQL是数据分析中使用最广泛的语言。Flink的Table API和SQL使用户能够用更少的时间和精力定义高效的流分析应用程序。此外,Flink Table API和SQL得到了有效的优化,它集成了大量的查询优化和优化的运算符实现。但并非所有优化都是默认启用的,因此对于某些工作负载,可以通过启用某些选项来提高性能。

5.2 DataStream / DataSet API & Table API & SQL 之间的关系

如下图所示,现在从下往上我们逐层介绍。
《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门,学习,flink,sql

  • 最低级别的抽象只是提供有状态和及时的流处理。它通过Process Function嵌入到DataStream API中。它允许用户自由处理来自一个或多个流的事件,并提供一致的容错状态。此外,用户可以注册事件时间和处理时间回调,使程序能够实现复杂的计算。
  • 在实践中,许多应用程序不需要上面描述的低级抽象,而是可以根据核心API进行编程:DataStream API(有界/无界流)。这些流畅的API为数据处理提供了通用的构建块,如各种形式的用户指定的转换、联接、聚合、窗口、状态等。在这些API中处理的数据类型在各自的编程语言中表示为类。
    低级别的Process Function与DataStream API集成在一起,从而可以按需使用低级别的抽象。数据集API在有界数据集上提供了额外的基元,如循环/迭代。
  • Table API是一个以表为中心的声明性DSL,表可以是动态变化的表(当表示流时)。Table API遵循(扩展的)关系模型:表附加了一个模式(类似于关系数据库中的表),API提供了类似的操作,如选择、项目、联接、分组传递、聚合等。Table API程序以声明的方式定义了应该执行的逻辑操作,而不是确切地指定操作代码的外观。虽然Table API可以通过各种类型的用户定义函数进行扩展,但它的表达能力不如Core API,使用起来更简洁(编写的代码更少)。此外,Table API 程序还通过一个优化器,该优化器在执行之前应用优化规则。
    可以在表和数据流/数据集之间无缝转换,允许程序将表API与数据流和数据集API混合。
  • Flink提供的最高级别抽象是SQL。这种抽象在语义和表达上都类似于Table API,但将程序表示为SQL查询表达式。SQL抽象与Table API密切交互,SQL查询可以在Table API中定义的表上执行。

5.3 Flink Table API 添加依赖

在前面的例子中,我们已经添加了 flink-clients 核心依赖,现在使用Table API 时,需要额外添加两个依赖,如下所示:

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner_${scala.binary.version}</artifactId>
    <version>${flink.version}</version>
</dependency>

其中,${flink.version} 为 1.14.6 ,而 {scala.binary.version} 为 2.11。

5.4 Flink Table API / SQL 第一个例子 StreamSQLExample

这个例子大概可以理解为:总共两个订单,每个订单里包含三条记录,总共六条记录。形成一张表,然后根据订单中 product 字段进行 UNION 操作,并把最终结果打印。

import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.util.Arrays;

/**
 * 摘录自 flink 1.14.6 源码例子
 * @author Smileyan
 */
public class StreamSQLExample {
    public static void main(String[] args) throws Exception {

        // set up the Java DataStream API
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // set up the Java Table API
        final StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        final DataStream<Order> orderA =
                env.fromCollection(
                        Arrays.asList(
                                new Order(1L, "beer", 3),
                                new Order(1L, "diaper", 4),
                                new Order(3L, "rubber", 2)));

        final DataStream<Order> orderB =
                env.fromCollection(
                        Arrays.asList(
                                new Order(2L, "pen", 3),
                                new Order(2L, "rubber", 3),
                                new Order(4L, "beer", 1)));

        // convert the first DataStream to a Table object
        // it will be used "inline" and is not registered in a catalog
        final Table tableA = tableEnv.fromDataStream(orderA);

        // convert the second DataStream and register it as a view
        // it will be accessible under a name
        tableEnv.createTemporaryView("TableB", orderB);

        // union the two tables
        final Table result =
                tableEnv.sqlQuery(
                        "SELECT * FROM "
                                + tableA
                                + " WHERE amount > 2 UNION ALL "
                                + "SELECT * FROM TableB WHERE amount < 2");

        // convert the Table back to an insert-only DataStream of type `Order`
        tableEnv.toDataStream(result, Order.class).print();

        // after the table program is converted to a DataStream program,
        // we must use `env.execute()` to submit the job
        env.execute();
    }

    // *************************************************************************
    //     USER DATA TYPES
    // *************************************************************************

    /** Simple POJO. */
    public static class Order {
        public Long user;
        public String product;
        public int amount;

        // for POJO detection in DataStream API
        public Order() {}

        // for structured type detection in Table API
        public Order(Long user, String product, int amount) {
            this.user = user;
            this.product = product;
            this.amount = amount;
        }

        @Override
        public String toString() {
            return "Order{"
                    + "user="
                    + user
                    + ", product='"
                    + product
                    + '\''
                    + ", amount="
                    + amount
                    + '}';
        }
    }
}

《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门,学习,flink,sql

5.5 Flink Table API / SQL 第二个例子 WordCountSQLExample

这个例子更加简单,因为连 union 的操作都已经省去了,直接从一个表中进行SELECT。

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.TableEnvironment;

/**
 *
 * @author Smileyan
 */
public class WordCountSQLExample {
    public static void main(String[] args) {

        // set up the Table API
        final EnvironmentSettings settings = EnvironmentSettings.newInstance().inBatchMode().build();
        final TableEnvironment tableEnv = TableEnvironment.create(settings);

        // execute a Flink SQL job and print the result locally
        tableEnv.executeSql(
                        // define the aggregation
                        "SELECT word, SUM(frequency) AS `count`\n"
                                // read from an artificial fixed-size table with rows and columns
                                + "FROM (\n"
                                + "  VALUES ('Hello', 1), ('Ciao', 1), ('Hello', 2)\n"
                                + ")\n"
                                // name the table and its columns
                                + "AS WordTable(word, frequency)\n"
                                // group for aggregation
                                + "GROUP BY word")
                .print();
    }
}

《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门,学习,flink,sql

5.6 参考资料

https://github.com/apache/flink/tree/release-1.14.6
https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/concepts/overview/

5.7 总结

第四章介绍了DataStream API和DataSet API的入门案例,而第五章则开始介绍了Table API以及基于其上的高层应用Flink SQL的基础知识。

在5.1中,阐述了Flink提供的两个关系API——Table API和SQL,用于统一流和批处理。无论是处理连续的流数据还是有界的批处理数据,在这两个接口中指定的查询具有相同的语义和结果。Table API和SQL接口与DataStream API无缝集成,用户可以轻松在它们之间切换。

5.2详细描述了DataStream、DataSet、Table API以及SQL之间的关系。从最低级别的抽象开始,介绍了DataStream API的Process Function,然后是DataStream API和DataSet API的一般构建块,最后到以表为中心的声明性DSL——Table API。最高级别的抽象是SQL,与Table API密切交互,允许通过SQL查询表达式执行操作。

在5.3中,介绍了Flink Table API的添加依赖,以及相应的Maven配置。

最后,在5.4和5.5中给出了两个Flink Table API / SQL的例子。StreamSQLExample展示了使用Table API和SQL进行流处理的例子,而WordCountSQLExample则展示了一个简单的批处理Flink SQL作业。

总体而言,本章深入介绍了Flink的Table API和SQL,以及它们与DataStream和DataSet API的关系,为使用Flink进行流和批处理提供了全面的基础知识。

Smileyan
2023.12.18 23:14文章来源地址https://www.toymoban.com/news/detail-778379.html

到了这里,关于《十堂课学习 Flink》第五章:Table API 以及 Flink SQL 入门的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

本文来自互联网用户投稿,该文观点仅代表作者本人,不代表本站立场。本站仅提供信息存储空间服务,不拥有所有权,不承担相关法律责任。如若转载,请注明出处: 如若内容造成侵权/违法违规/事实不符,请点击违法举报进行投诉反馈,一经查实,立即删除!

领支付宝红包 赞助服务器费用

相关文章

  • 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及JDBC示例(4)

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月10日
    浏览(39)
  • 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Elasticsearch示例(2)

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月11日
    浏览(50)
  • 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Kafka示例(3)

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月10日
    浏览(55)
  • 16、Flink 的table api与sql之连接外部系统: 读写外部系统的连接器和格式以及Apache Hive示例(6)

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年02月11日
    浏览(52)
  • 【flink番外篇】9、Flink Table API 支持的操作示例(1)-通过Table API和SQL创建表

    一、Flink 专栏 Flink 专栏系统介绍某一知识点,并辅以具体的示例进行说明。 1、Flink 部署系列 本部分介绍Flink的部署、配置相关基础内容。 2、Flink基础系列 本部分介绍Flink 的基础部分,比如术语、架构、编程模型、编程指南、基本的datastream api用法、四大基石等内容。 3、

    2024年01月17日
    浏览(58)
  • Flink Table API 与 SQL 编程整理

    Flink API 总共分为 4 层这里主要整理 Table API 的使用 Table API 是流处理和批处理通用的关系型 API , Table API 可以基于流输入或者批输入来运行而不需要进行任何修改。 Table API 是 SQL 语言的超集并专门为 Apache Flink 设计的, Table API 是 Scala 和 Java 语言集成式的 API 。与常规 SQL 语言

    2024年02月04日
    浏览(51)
  • Flink Table API/SQL 多分支sink

    在某个场景中,需要从Kafka中获取数据,经过转换处理后,需要同时sink到多个输出源中(kafka、mysql、hologres)等。两次调用execute, 阿里云Flink vvr引擎报错: 使用 StreamStatementSet. 具体参考官网: https://nightlies.apache.org/flink/flink-docs-release-1.15/zh/docs/dev/table/data_stream_api/#converting-betwe

    2024年02月11日
    浏览(97)
  • 【Flink SQL】Flink SQL 基础概念(一):SQL & Table 运行环境、基本概念及常用 API

    《 Flink SQL 基础概念 》系列,共包含以下 5 篇文章: Flink SQL 基础概念(一):SQL Table 运行环境、基本概念及常用 API Flink SQL 基础概念(二):数据类型 Flink SQL 基础概念(三):SQL 动态表 连续查询 Flink SQL 基础概念(四):SQL 的时间属性 Flink SQL 基础概念(五):SQL 时区问

    2024年03月21日
    浏览(72)
  • Flink(十五)【Flink SQL Connector、savepoint、CateLog、Table API】

           今天一天争取搞完最后这一部分,学完赶紧把 Kafka 和 Flume 学完,就要开始做实时数仓了。据说是应届生得把实时数仓搞个 80%~90% 才能差不多找个工作,太牛马了。         之前我们已经用过了一些简单的内置连接器,比如 \\\'datagen\\\' 、\\\'print\\\' ,其它的可以查看官网:

    2024年01月24日
    浏览(52)
  • Flink系列Table API和SQL之:时间属性

    基于时间的操作(比如时间窗口),需要定义相关的时间语义和时间数据来源的信息。在Table API和SQL中,会给表单独提供一个逻辑上的时间字段,专门用来在表处理程序中指示时间。 所谓的时间属性(time attributes),就是每个表模式结构(schema)的一部分。可以在创建表的DDL里直接定

    2023年04月09日
    浏览(43)

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

请作者喝杯咖啡吧~博客赞助

支付宝扫一扫领取红包,优惠每天领

二维码1

领取红包

二维码2

领红包