当前位置： 首页 > article >正文

大数据-132 - Flink SQL 基本介绍 与 HelloWorld案例

article 2024/11/11 13:16:00

点一下关注吧！！！非常感谢！！持续更新！！！

目前已经更新到了：

  • Hadoop（已更完）
  • HDFS（已更完）
  • MapReduce（已更完）
  • Hive（已更完）
  • Flume（已更完）
  • Sqoop（已更完）
  • Zookeeper（已更完）
  • HBase（已更完）
  • Redis （已更完）
  • Kafka（已更完）
  • Spark（已更完）
  • Flink（正在更新！）

章节内容

上节我们完成了如下的内容：

  • Flink CEP 案例
  • 检测交易活跃用户
  • 超时未交付

在这里插入图片描述

Flink SQL

Flink SQL 是 Apache Flink 提供的一种高层次的查询语言接口，它基于 SQL 标准，为开发者提供了处理流式数据和批处理数据的能力。Flink SQL 允许用户使用标准 SQL 查询语言在数据流和数据表上执行复杂的操作，适用于多种应用场景，如实时分析、数据流处理、机器学习等。下面是 Flink SQL 的一些重要概念和功能：

流与批统一的查询模式

Flink SQL 的一大特点是流处理和批处理的统一性。通过同一套 SQL 语法，用户可以同时处理静态数据（批处理）和动态数据（流处理）。这使得应用程序的开发更加简化，因为可以用相同的逻辑编写实时流数据处理和历史数据的查询。

动态表 (Dynamic Tables)

Flink SQL 通过动态表的概念将流数据建模为不断变化的表。这种动态表随着时间推移不断更新，数据的每个变化（插入、更新、删除）都会影响表的状态。通过动态表的概念，Flink 可以使用 SQL 查询连续的流数据，并在查询执行时获得不断更新的结果。

窗口操作 (Windowing)

在流式数据处理场景中，窗口操作非常重要。Flink SQL 提供了多种类型的窗口操作，包括：

  • 滚动窗口 (Tumbling Window)：将数据按照固定长度分割成不重叠的窗口。
  • 滑动窗口 (Sliding Window)：窗口之间存在重叠，数据可能被分配到多个窗口。
  • 会话窗口 (Session Window)：窗口由活动间隔定义，不同的事件可能会聚合在一个窗口中。

连接操作 (Joins)

Flink SQL 支持多种连接操作：

  • 流与流的连接：允许用户将多个流结合在一起，基于时间或键进行匹配。
  • 流与表的连接：将静态表与流数据进行匹配，从而使流式数据处理能够结合历史数据或参考数据。
  • 时态表连接 (Temporal Table Join)：用于将流数据与一个时态表进行连接，时态表会随着时间不断更新。

内置函数和自定义函数

Flink SQL 提供了丰富的内置函数，涵盖了字符串操作、数学运算、时间日期处理、聚合操作等。此外，Flink SQL 还支持用户自定义函数（UDF、UDTF、UDAF），用户可以根据具体需求扩展 SQL 的功能。

Table API 与 SQL API 的互操作性

Flink 提供了两种高级数据处理 API：

  • Table API：一种与关系代数类似的编程接口，支持链式调用，功能类似于 SQL。
  • SQL API：用户可以直接使用标准 SQL 语句进行数据处理。

Table API 和 SQL API 具有很高的互操作性，用户可以在同一个程序中混合使用这两者。例如，可以先用 Table API 进行表定义和部分操作，再通过 SQL 语句执行复杂的查询。

支持多种数据源和数据接收器

Flink SQL 支持连接多种数据源和数据接收器，如 Kafka、文件系统、数据库（如 MySQL、PostgreSQL）、Hive、HBase 等。通过 SQL 语法，用户可以轻松地将流数据写入这些外部系统，也可以从这些系统中读取数据进行处理。

状态管理与容错机制

Flink SQL 继承了 Flink 强大的状态管理和容错机制。在流处理任务中，Flink SQL 能够有效地处理有状态的计算，并保证在失败时自动恢复。基于 Flink 的检查点（Checkpointing）和保存点（Savepoint）机制，Flink SQL 提供了 Exactly-Once 的状态一致性保障。

实时分析与 ETL

Flink SQL 可以用于实时数据的分析与处理，常用于构建实时 ETL (Extract, Transform, Load) 流程。例如，用户可以通过 SQL 查询对从 Kafka、数据库等数据源接收到的流数据进行清洗、过滤、转换，并将结果写入到其他系统中（如 Elasticsearch、HDFS、JDBC）。

HelloWorld

添加依赖

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table</artifactId>
    <type>pom</type>
    <version>${flink.version}</version>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-api-java-bridge_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-table-planner-blink_2.12</artifactId>
    <version>${flink.version}</version>
    <scope>provided</scope>
</dependency>

依赖说明：

  • flink-table-api-java-bridge_2.12：桥接器，主要负责 TableAPI 和 DataStream/DataSetAPI 的连接支持，按照语言分Java和Scala。
  • flink-table-planner-blink_2.12：计划期，是TableAPI最主要的部分，提供了运行时环境和生成程序执行计划的Planner。
  • 如果是生产环境，则已经有 planner，就只需要有bridge就可以了
  • flink-table：基础依赖

编写代码

package icu.wzk;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;


public class TableApiDemo {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        DataStreamSource<Tuple2<String, Integer>> data = env.addSource(new SourceFunction<Tuple2<String, Integer>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Integer>> ctx) throws Exception {
                while (true) {
                    ctx.collect(new Tuple2<>("name", 10));
                    Thread.sleep(1000);
                }
            }

            @Override
            public void cancel() {

            }
        });

        // =======================
        // Table 方式
        Table table = tableEnvironment.fromDataStream(data, $("name"), $("age"));
        // 对Table的数据查询
        Table name = table.select($("name"));
        // 将数据输出到控制台
        DataStream<Tuple2<Boolean, Row>> result = tableEnvironment.toRetractStream(name, Row.class);
        result.print();
        System.out.println("=========================");
        // =======================
        // SQL 方式
        tableEnvironment.createTemporaryView("users",data, $("name"), $("age"));
        String sql = "select name from users";
        table = tableEnvironment.sqlQuery(sql);
        result = tableEnvironment.toRetractStream(table, Row.class);
        result.print();
        System.out.println("=========================");
        env.execute("TableApiDemo");
    }

}

运行代码

控制台会一直不间断的输出如下的内容：

=========================
=========================
1> (true,name)
6> (true,name)
2> (true,name)
7> (true,name)
3> (true,name)
8> (true,name)
4> (true,name)
1> (true,name)
5> (true,name)
2> (true,name)
6> (true,name)
3> (true,name)

控制台的运行结果如下所示：
在这里插入图片描述


http://www.kler.cn/a/305328.html

相关文章：

  • 小程序源码-模版 100多套小程序（附源码）
  • 2024 第五次周赛
  • BEAGLE: Forensics of Deep Learning Backdoor Attack for Better Defense（论文阅读）
  • Python操作系统交互：subprocess库的基本应用
  • 什么是直方图算法
  • 如何利用 Python 的爬虫技术获取淘宝天猫商品的价格信息？
  • 【LLM多模态】CogVideoX文生视频模型结构和训练过程
  • 【openGauss】检查工具gs_check,gs_checkperf的应用
  • Redisson 异步释放锁
  • 什么是485无线通信
  • 磁盘写操作压力测试工具的设计与实现
  • ChatGLM3模型搭建教程
  • OpenCV结构分析与形状描述符（20）计算一个包围给定点集的最小外接圆函数minEnclosingCircle()的使用
  • 研1日记12
  • C++系列-函数对象/仿函数
  • [网络]TCP/IP协议 之 网络层IP协议(3)
  • 艾丽卡的区块链英语小课堂
  • 基于ssm+vue+uniapp的新生报到系统小程序
  • 第4步CentOS配置SSH服务用SSH终端XShell等连接方便文件上传或其它操作
  • 实战外网配置——光猫桥接+路由器PPPoE拨号+防火墙外网链路健康检查+外网流量负载均衡
  • kafka mirror maker之实现两个kafka集群之间的数据同步
  • HTML + CSS - 网页布局之一般布局浮动布局
  • JVM 体系与结构
  • Node.js 中间件与洋葱模型
  • SonicWall SSL VPN曝出高危漏洞，可能导致防火墙崩溃
  • java --- 性能优化01