首页 > 数据库 >10分钟了解Flink SQL使用

10分钟了解Flink SQL使用

时间：2024-05-12 09:10:06浏览次数：23  
标签：10 Flink 批处理 flink SQL 流批 csv

Flink 是一个流处理和批处理统一的大数据框架，专门为高吞吐量和低延迟而设计。开发者可以使用SQL进行流批统一处理，大大简化了数据处理的复杂性。本文将介绍Flink SQL的基本原理、使用方法、流批统一，并通过几个例子进行实践。

1、Flink SQL基本原理

Flink SQL建立在Apache Flink之上，利用Flink的强大处理能力，使得用户可以使用SQL语句进行流数据和批数据处理。Flink SQL既支持实时的流数据处理，也支持有界的批数据处理。

Flink SQL用SQL作为处理数据的接口语言，将SQL语句转换成数据流图（Dataflow Graph），再由Flink引擎执行。

2、Flink SQL固定编码套路

使用Flink SQL时，我们通常会遵循如下编码套路，这些套路和使用Flink API的套路是一样的：

  • 环境准备：初始化一个TableEnvironment对象，它是执行Flink SQL语句的核心。这个环境可以是流数据环境，也可以是批数据环境。
  • 数据源定义：通过CREATE TABLE语句定义输入数据源（source），可以是Kafka、CSV文件等。
  • 数据处理：编写SQL语句对数据进行处理，如查询、过滤、聚合等。
  • 数据输出：通过CREATE TABLE定义输出数据源（sink），并将处理结果输出。

3、Flink SQL代码示例

以下是一个从CSV文件读取数据，通过SQL查询，再将数据输出到CSV的完整例子。

  • 先准备input.csv文件内容，如下：
1,product_A,10.5
2,product_B,20.3
3,product_C,15.8
1,product_D,12.2
2,product_A,18.7
  • 编写demo代码

编写代码之前先在pom.xml中添加依赖：

<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-csv</artifactId>
    <version>${flink.version}</version>
</dependency>
<dependency>
    <groupId>org.apache.flink</groupId>
    <artifactId>flink-json</artifactId>
    <version>${flink.version}</version>
</dependency>

示例代码如下：

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class FlinkSqlDemo {
    public static void main(String[] args) throws Exception {
        // 设置环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1); //为了方便测试看效果，这里并行度设置为1
        // 使用EnvironmentSettings创建StreamTableEnvironment，明确设置为批处理模式
        EnvironmentSettings settings = EnvironmentSettings
                .newInstance()
                .inBatchMode() // 设置为批处理模式，这样后续才能一次性的输出到csv中
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // 定义输入数据源
        String createSourceTableDdl = "CREATE TABLE csv_source (" +
                " user_id INT," +
                " product STRING," +
                " order_amount DOUBLE" +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'file:///path/input.csv'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createSourceTableDdl);

//        // 编写 SQL 查询
//        String query = "SELECT user_id, SUM(order_amount) AS total_amount FROM csv_source GROUP BY user_id";
//        // 执行查询并打印
//        tableEnv.executeSql(query).print();
//        env.execute("Flink SQL Demo");

        // 定义输出数据源
        String createSinkTableDdl = "CREATE TABLE csv_sink (" +
                " user_id INT," +
                " total_amount DOUBLE" +
                ") WITH (" +
                " 'connector' = 'filesystem'," +
                " 'path' = 'file:///path/output.csv'," +
                " 'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createSinkTableDdl);

        // 执行查询并将结果输出到csv_sink
        String query = "INSERT INTO csv_sink " +
                "SELECT user_id, SUM(order_amount) as total_amount " +
                "FROM csv_source " +
                "GROUP BY user_id";
        tableEnv.executeSql(query);
//        env.execute("Flink SQL Job");
    }
}
  • 执行结果如下：

4、Flink SQL做流批统一

什么是流批统一？

流批统一是大数据处理领域的一个概念，它指的是使用一套代码来同时处理流数据（Streaming）和批数据（Batching）。

流处理和批处理的区别如下：

  1. 批处理（Batch Processing）：
    • 批处理是指在某一时间点处理大量数据的手段。
    • 它通常涉及到对大量静止的（不再变化的）数据集进行一次性的处理。
    • 批处理作业通常在数据集完整可用后开始执行，并且经常是在数据仓库中进行。
    • 例如，一个电商平台可能在一天结束时运行一个批处理作业来处理当天所有的交易记录。
  2. 流处理（Stream Processing）：
    • 流处理是指对数据实时进行处理，通常是数据生成或接收的同时立即进行。
    • 流处理适用于连续的数据输入，这些数据一直在变化，需要立即响应。
    • 例如，社交媒体平台在接收到新的帖子时，可能会实时分析这些帖子的内容和流行趋势。

在早期，流处理和批处理通常需要不同的系统来执行。对于批处理，可能使用如Hadoop这样的框架；而对于流处理，可能使用如Apache Storm这样的框架。这就导致开发者要同时学习多种框架才能处理不同类型的数据作业。

流批统一的概念，就是将这两种数据处理方式合并到一个平台中，这样一个系统既可以处理静止的大批量数据集，也可以处理实时的数据流。这样做的优点是显而易见的：

  • 统一的API：开发人员只需要学习和使用一套工具和API，可以共享更多的代码和逻辑。
  • 维护简便：只需维护一个系统，可以减少学习成本，减轻运维压力，减少故障点。
  • 灵活的数据处理：可以根据不同的业务需求灵活选择数据处理方式。

Flink SQL流批一体的实现原理

Flink很好的实现了流批统一，可以让开发人员用相同的方式来编写批处理和流处理程序。不论是对有界（批处理）还是无界（流处理）的数据源，Flink都可以使用相同的API和处理逻辑来处理数据。

Flink 通过内置的表抽象来实现流批一体，这里的"表"可以是动态变化的（例如，来自实时数据流的表）或是静态的（例如，存储在文件或数据库中的批量数据表）。Flink SQL引擎会根据数据的实际来源自动优化执行计划。

Flink SQL的流批统一核心在于三点：

  • 统一的API和SQL语义：Flink SQL提供一致的查询构建块（如窗口、时间处理函数），这些在流处理和批处理中语义一致，确保不同模式下行为的统一性。
  • 透明的状态处理：无论是流处理还是批处理，Flink都能够保持和恢复状态，为开发者提供一致的高容错性体验。
  • 多模态存储和处理能力：Flink SQL能够访问不同存储介质的数据，这意味着相同的SQL语句可以无缝在流数据和存储的批量数据上执行。

Flink SQL流批统一的代码示例

以下是一个完整的代码示例，用Flink来实现流批统一处理。Flink同时从Kafka 和 CSV读取数据，然后合并查询再输出结果：

  • 代码示例

代码中，先配置了Flink的流处理环境和表环境，然后用DDL语句在Flink中注册了Kafka和文件系统数据源。接着执行了一个SQL查询来合并来自这两种数据源的数据，并计算总金额。最后，打印出查询结果并开始执行Flink作业。

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

public class StreamBatchUnifiedDemo {
    public static void main(String[] args) throws Exception {
        // 设置流处理的环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        EnvironmentSettings settings = EnvironmentSettings.newInstance()
                .inStreamingMode()
                .build();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env, settings);

        // Kafka 流处理表
        String createKafkaSourceDDL = "CREATE TABLE kafka_stream_orders (" +
                "order_id STRING," +
                "amount DOUBLE)" +
                "WITH (" +
                "'connector' = 'kafka'," +
                "'topic' = 'topic_test'," +
                "'properties.bootstrap.servers' = '10.20.1.26:9092'," +
                "'format' = 'json'," +
                "'scan.startup.mode' = 'latest-offset'" +
                ")";
        tableEnv.executeSql(createKafkaSourceDDL);

        // 文件系统批处理表
        String createFilesystemSourceDDL = "CREATE TABLE file_batch_orders (" +
                "order_id STRING," +
                "amount DOUBLE)" +
                "WITH (" +
                "'connector' = 'filesystem'," +
                "'path' = 'file:///Users/yclxiao/Project/bigdata/flink-blog/doc/input_order.csv'," +
                "'format' = 'csv'" +
                ")";
        tableEnv.executeSql(createFilesystemSourceDDL);

        // 执行统一查询，计算总金额
        Table resultTable = tableEnv.sqlQuery("SELECT SUM(amount) FROM (" +
                "SELECT amount FROM kafka_stream_orders " +
                "UNION ALL " +
                "SELECT amount FROM file_batch_orders)");

        // 打印结果
        tableEnv.toRetractStream(resultTable, Row.class).print();

        // 开始执行程序
        env.execute("Stream-Batch Unified Job");
    }
}
  • 执行效果

通过以上示例代码，可以看出Flink SQL的流批一体设计：相同的SQL语句可以用在流处理和批处理中，而不需要做任何修改。Flink背后的执行引擎会自动根据数据的特性（流或者批）来进行相应的优化执行。

这就是Flink SQL非常强大的地方，它减少了开发者需要写不同代码逻辑的需求，简化了复杂的数据处理流程。

5、总结

Flink SQL是一个非常强大的数据处理工具，可以应对多种复杂的数据处理场景。

本文主要介绍了Flink SQL的基本原理、编码套路、流批统一，再结合正确的代码示例进行实践。希望对你有帮助。

完整代码地址：https://github.com/yclxiao/flink-blog

======>>>>>> 关于我 <<<<<<======

本篇完结！欢迎点赞 关注 收藏！！！

原文链接：https://mp.weixin.qq.com/s/WqyCjiIMK49T6eDl_Upc1A

标签：10,Flink,批处理,flink,SQL,流批,csv
From： https://www.cnblogs.com/mangod/p/18187474

相关文章

  • SystemVerilog -- 10.1 SystemVerilog Covergroup and Coverpoint
    SystemVerilogCovergroupandCoverpointSystemVerilog是一种用户定义的类型，用于封装覆盖率模型的规范。它们可以定义一次，并通过函数在不同位置实例化多次。covergroupnewcovergroup可以在包、模块、程序、接口类中定义，通常封装以下信息：AsetofcoveragepointsCrosscov......
  • java MySQL的in查询两个字段同时相等的多个数据查询
    在Java中想进行下面的查询，不想写循环一条条查selectid,name,address,age fromuserswherename='A' and address='addr1'selectid,name,address,age fromuserswherename='B' and address='addr2'selectid,name,address,age fromusers......
  • SystemVerilog -- 10.0 SystemVerilog Functional Coverage
    SystemVerilogFunctionalCoverageWhatisfunctionalcoverage?functionalcoverage是测试对设计的哪些功能/特性的衡量。这在约束随机验证（CRV）中非常有用，可以了解回归中的一组测试涵盖了哪些特征。Whatareitslimitations?这仅与为它编写的代码一样好。假设您在设计文档......
  • 前端使用 Konva 实现可视化设计器（10）- 对齐线
    请大家动动小手，给我一个免费的Star吧~大家如果发现了Bug，欢迎来提Issue哟~github源码gitee源码示例地址不知不觉来到第10章了，感觉接近尾声了。。。对齐线先看效果：这里交互有两个部分：1、节点之间的对齐线2、对齐磁贴多选的情况下，效果是一样的：主要逻辑会放在......
  • 贝壳面试：MySQL联合索引，最左匹配原则是什么？
    文章很长，且持续更新，建议收藏起来，慢慢读！疯狂创客圈总目录博客园版为您奉上珍贵的学习资源：免费赠送:《尼恩Java面试宝典》持续更新+史上最全+面试必备2000页+面试必备+大厂必备+涨薪必备免费赠送:《尼恩技术圣经+高并发系列PDF》，帮你实现技术自由，完成职业升级，薪......
  • 敏捷冲刺-5月10日
    敏捷冲刺-Day-05所属课程软件工程2024作业要求团队作业4—项目冲刺作业目标完成第5篇Scrum冲刺博客冲刺日志集合贴https://www.cnblogs.com/YXCS-cya/p/181788031.项目燃尽图1.1第五日-5月10日进度当前进度逐渐加快2.会议记录2.1会议主题第5......
  • 关于VHDL中Loop State error...loop must terminate within 10,000 iterations错误解
    关于VHDL中LoopStateerror...loopmustterminatewithin10,000iterations错误解决方法首先比较下面两段代码：(使用while循环描述偶校验位产生电路)代码一：libraryieee;useieee.std_logic_1164.all;useieee.std_logic_unsigned.all;useieee.std_logic_arith.all;ent......
  • MySQL中的一些其他比较符
    一、在两者之间（可以看成数学中的闭区间）BETWEEN...AND...在两个值之间（包含）SELECTLAST_NAME,SALARYFROMemployeesWHERESALARYBETWEEN3500AND6000;二、匹配一个任意值列表IN（set）使用该关键字之后会给定一个列表，所取的值都在这个列表中间，在查询数据的时候，会对这个set中......
  • SQL SERVER 数据库清空语句 忽略外键 触发器 等（转载）
    有时候我们想清空SQLServer中所有表的数据，但是由于有外键约束和触发器，有时候清表语句无法执行，下面的语句可以关掉和开启一个数据库中的所有外键约束和触发器，以便执行清表语句：USEClothesShopEXECUTEsp_MSForEachTable'ALTERTABLE?NOCHECKCONSTRAINTALL'EXECUTEsp_MS......
  • 专业商用远程控制软件，低至10元每月
    如今不论工作还是生活，远程控制软件已成为我们的必备工具。说到远程软件，市面上有很多款。但“个人免费版”，往往会有限速、卡顿、不清晰等问题；商业版很强大，但普遍价格昂贵。那么有没有既便宜又专业的商业级远控软件呢？现在，它来了，它来了！Splashtop家族的商业版远程软件BusinessAcc......

赞助商

阅读排行

网站主页关于我们联系我们网站地图

本网站内容转载自其他媒体，侵权联系[admin##ips99.com]。

Copyright © 2020-2023 IPS99 版权所有 IPS99