Flink-Table API 实践编程 StreamTableEnvironment（九）下

最新推荐文章于 2024-05-21 02:33:07 发布
最新推荐文章于 2024-05-21 02:33:07 发布
阅读量1.5k 收藏 1
点赞数 2
分类专栏： flink 文章标签： flink
版权声明：本文为博主原创文章，遵循 CC 4.0 BY-SA 版权协议，转载请附上原文出处链接和本声明。
本文链接：https://blog.csdn.net/springk/article/details/104707260
版权

这一节主要是实践编程StreamTableEnvironment下相关table api的使用信息，代码中模拟输入流采用的是socket数据流输入模式。
实例一：

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

public class FlinkTableApiStreamingExample {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //source,这里使用socket连接获取数据
        DataStreamSource<String> text = env.socketTextStream("127.0.0.1", 9999, "\n");

        //处理输入数据流，转换为StudentInfo类型，方便后续处理
        SingleOutputStreamOperator<StudentInfo> dataStreamStudent = text.flatMap(new FlatMapFunction<String, StudentInfo>() {
            @Override
            public void flatMap(String s, Collector<StudentInfo> collector){
                String infos[] = s.split(",");
                if(StringUtils.isNotBlank(s) && infos.length==5){
                    StudentInfo studentInfo = new StudentInfo();
                    studentInfo.setName(infos[0]);
                    studentInfo.setSex(infos[1]);
                    studentInfo.setCourse(infos[2]);
                    studentInfo.setScore(Float.parseFloat(infos[3]));
                    studentInfo.setTimestamp(Long.parseLong(infos[4]));
                    collector.collect(studentInfo);
                }
            }
        });

        //注册dataStreamStudent流到表中，表名为：studentInfo
        tEnv.registerDataStream("studentInfo",dataStreamStudent,"name,sex,course,score,timestamp");

        //GroupBy Aggregation 根据name分组，统计学科数量
        Table counts = tEnv.scan("studentInfo")
                .groupBy("name")
                .select("name, course.count as cnt");
        DataStream<Tuple2<Boolean, Row>> resultCountsAggr = tEnv.toRetractStream(counts, Row.class);
        resultCountsAggr.print();

        //GroupBy Aggregation distinct 根据name分组，统计学科数量
        Table groupByDistinctResult = tEnv.scan("studentInfo")
                .groupBy("name")
                .select("name, score.sum.distinct as d");
        DataStream<Tuple2<Boolean, Row>> resultDistinctAggr = tEnv.toRetractStream(groupByDistinctResult, Row.class);
        resultDistinctAggr.print();

        env.execute("studentScoreAnalyse");

    }
}

输入数据信息如下：
在这里插入图片描述
返回结果信息如下：
1、第一个返回数据如下
在这里插入图片描述
2、第二个返回结果如下：在这里插入图片描述
实例二：

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.table.api.Over;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.Tumble;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.types.Row;
import org.apache.flink.util.Collector;

import javax.annotation.Nullable;

public class FlinkTableApiStreamingWatermarkExample {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        //source,这里使用socket连接获取数据
        DataStreamSource<String> text = env.socketTextStream("127.0.0.1", 9999, "\n");

        //处理输入数据流，转换为StudentInfo类型，方便后续处理
        SingleOutputStreamOperator<StudentInfo> dataStreamStudent = text.flatMap(new FlatMapFunction<String, StudentInfo>() {
            @Override
            public void flatMap(String s, Collector<StudentInfo> collector){
                String infos[] = s.split(",");
                if(StringUtils.isNotBlank(s) && infos.length==5){
                    StudentInfo studentInfo = new StudentInfo();
                    studentInfo.setName(infos[0]);
                    studentInfo.setSex(infos[1]);
                    studentInfo.setCourse(infos[2]);
                    studentInfo.setScore(Float.parseFloat(infos[3]));
                    studentInfo.setTimestamp(Long.parseLong(infos[4]));
                    collector.collect(studentInfo);
                }
            }
        });


        //以下实例采用时间窗口模式，需要设置时间属性，否则代码报错
        //EventTime
        DataStream<StudentInfo> dataStream = dataStreamStudent.assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<StudentInfo>() {
            private final long maxTimeLag = 5000; // 5 seconds

            @Nullable
            @Override
            public Watermark getCurrentWatermark() {
                return new Watermark(System.currentTimeMillis() - maxTimeLag);
            }

            @Override
            public long extractTimestamp(StudentInfo studentInfo, long l) {
                return studentInfo.getTimestamp();
            }
        });

        //注册dataStreamStudent流到表中，表名为：studentInfo
        Table tableEvent = tEnv.fromDataStream(dataStream, "name,sex,course,score,timestamp.rowtime");

        //GroupBy Window
        Table resultGroupByWindow = tableEvent
                .filter("name.isNotNull && course.isNotNull ")
//                .select("name.lowerCase() as name, course, utc2local(timestamp) as timestamp")
                .window(Tumble.over("1.minutes").on("timestamp").as("hourlyWindow"))
                .groupBy("hourlyWindow, name, course")
                .select("name, hourlyWindow.end, hourlyWindow.start,hourlyWindow.rowtime as hour, course, course.count as courseCount");

        DataStream<Row> result2 = tEnv.toAppendStream(resultGroupByWindow, Row.class);
        result2.print();

        //GroupBy Window Over
        Table resultOverWindow = tableEvent
                .window(Over
                        .partitionBy("name")
                        .orderBy("timestamp")
                        .preceding("1.minutes")
                        .following("CURRENT_RANGE")
                        .as("w"))
                .select("name, score.avg over w,score.max over w, score.min over w"); // sliding aggregate

        DataStream<Row> resultOver = tEnv.toAppendStream(resultOverWindow, Row.class);
        resultOver.print();

        // Distinct aggregation on time window group by  BatchTableEnvironment不支持
        Table groupByWindowDistinctResult = tableEvent
                .window(Tumble.over("1.minutes").on("timestamp").as("w")).groupBy("name,w")
                .select("name, score.sum.distinct as d");
        DataStream<Row> resultDistinct = tEnv.toAppendStream(groupByWindowDistinctResult, Row.class);
        resultDistinct.print();

//
        // Distinct aggregation on over window  TODO
        Table resultOverWindowDistinct = tableEvent
                .window(Over
                        .partitionBy("name")
                        .orderBy("timestamp")
                        .preceding("1.minutes")
                        .as("w"))
                .select("name, score.sum.distinct over w, score.max over w, score.min over w");

        env.execute("studentScoreAnalyse");

    }
}

输入数据信息：
在这里插入图片描述
输出数据信息：
在这里插入图片描述

StreamTableEnvironment 源码详解及应用示例
2301_79367634的博客
08-16 431
StreamTableEnvironment 是 Flink Table API 和 SQL 的核心入口之一，它允许用户在实时数据流中执行 SQL 查询、聚合操作以及表与流之间的转换。在分析实时数据时，StreamTableEnvironment 提供了一套强大的功能，使得开发者可以使用熟悉的 SQL 查询语言进行数据处理和分析。
95-110-040-源码-Env-StreamTableEnvironment
九师兄
07-03 721
1.视界 # 2.概述 TableEnvironment是Table API和SQL集成的核心概念，他主要负责： 1. 在Table内部目录中注册表 2. 注册一个外部表 3. 执行SQL查询 4. 注册用户行为的（标量，表或聚合）函数 5. 将DataStream或者DataSet转换为Table 6. 持有对ExecutionEnvironment或的引用StreamExecutionEnvi...
Flink-Table StreamTableEnvironment基础知识（三）
springk的专栏
01-16 9403
StreamTableEnvironment用于流计算场景，流计算的对象是DataStream。相比 TableEnvironment，StreamTableEnvironment 提供了 DataStream 和 Table 之间相互转换的接口，如果用户的程序除了使用 Table API & SQL 编写外，还需要使用到 DataStream API，则需要使用 StreamTableE...
Flink之StreamTableEnvironment对象
sijiwang95的博客
04-11 2074
1.executeSql("sql 语句")可以执行SQL#2.sqlQuery("sql 语句")执行SQL查询，返回查询结果加载table到内存中把结果插入到table中在内存中创建临时表。
Flink之Table API
01-20
Apache Flink 的 Table API 是一个强大的工具，专为处理流数据和批数据而设计，它提供了统一的编程模型，使得开发者无需对代码进行修改就能在流处理和批处理之间切换。Table API 是 SQL 语言的一个超集，尤其针对 ...
flink-Table-API
互联网知识分享
05-20 125
但要注意 此时select里面是要k v1 v2的格式，必须要要有操作，不能直接输出源数据，且由于做了groupBy的操作 所以 转换DataStream的时候，就不能使用toAppendStream 而是toRetractStream。：表示 指定的时间是哪一个字段。值得注意的是，Table API and SQL目前尚未完全完善，还在积极的开发中，所以并不是所有的算子操作都可以通过其实现。（4）动态表的查询与静态表一样，但是，在查询动态表的时候，SQL 会做连续查询，不会终止。而as 'win：表示。
大数据学习之Flink——15TableAPI 和 Flink SQL
bingque6535的博客
07-15 341
Table API 以及 SQL 能够统一处理批量和实时计算业务, 无须切换修改任何应用代码就能够基于同一套 API 编写流式应用和批量应用, 从而达到真正意义的批流统一 一. 开发环境构建 在使用 Table API 和 SQL 开发 Flink 应用之前, 通过添加 Maven 的依赖配置到项目中, 在本地工程中引入相应的依赖库, 库中包含了 TableAPI 和 SQL 接口。 <dependency> <groupId>org.apache.flink<.
Flink Table API & SQL编程指南(1)
jmx_bigdata的博客
06-01 785
Apache Flink提供了两种顶层的关系型API，分别为Table API和SQL，Flink通过Table API&SQL实现了批流统一。其中Table API是用于Scala和Java的语言集成查询API，它允许以非常直观的方式组合关系运算符（例如select，where和join）的查询。Flink SQL基于Apache Calcite 实现了标准的SQL，用户可以使用标准的SQL处理数据集。Table API和SQL与Flink的DataStream和DataSet API紧密集成在.
Flink Table原理与代码实例讲解
最新发布
AI天才研究院
05-21 544
1. 背景介绍 1.1 大数据处理的演变与挑战 随着互联网和移动互联网的快速发展，全球数据量呈现爆炸式增长，大数据处理技术也随之不断演进。从早期的 Hadoop 生态系统到 Spark、Flink 等新一代分布式计算引擎，大数据处理能力不断提升，但同时也面临着新的挑战：
大数据之flink教程-TableAPI和SQL.pdf
07-27
第一章 整体介绍 2 1.1 什么是 Table API 和 Flink SQL 2 1.2 需要引入的依赖 2 1.3 两种 planner（old & blink）的区别 4 第二章 API 调用 5 2.1 基本程序结构 5 2.2 创建表环境 5 2.3 在 Catalog 中注册表 7 2.3.1 表（Table）的概念 7 2.3.2 连接到文件系统（Csv 格式） 7 2.3.3 连接到 Kafka 8 2.4 表的查询 9 2.4.1 Table API 的调用 9 2.4.2 SQL 查询 10 2.5 将 DataStream 转换成表 11 2.5.1 代码表达 11 2.5.2 数据类型与 Table schema 的对应 12 2.6. 创建临时视图（Temporary View） 12 2.7. 输出表 14 2.7.1 输出到文件 14 2.7.2 更新模式（Update Mode） 15 2.7.3 输出到 Kafka 16 2.7.4 输出到 ElasticSearch 16 2.7.5 输出到 MySql 17 2.8 将表转换成 DataStream 18 2.9 Query 的解释和执行 20 1. 优化查询计划 20 2. 解释成 DataStream 或者 DataSet 程序 20 第三章 流处理中的特殊概念 20 3.1 流处理和关系代数（表，及 SQL）的区别 21 3.2 动态表（Dynamic Tables） 21 3.3 流式持续查询的过程 21 3.3.1 将流转换成表（Table） 22 3.3.2 持续查询（Continuous Query） 23 3.3.3 将动态表转换成流 23 3.4 时间特性 25 3.4.1 处理时间（Processing Time） 25 3.4.2 事件时间（Event Time） 27 第四章 窗口（Windows） 30 4.1 分组窗口（Group Windows） 30 4.1.1 滚动窗口 31 4.1.2 滑动窗口 32 4.1.3 会话窗口 32 4.2 Over Windows 33 1） 无界的 over window 33 2） 有界的 over window 34 4.3 SQL 中窗口的定义 34 4.3.1 Group Windows 34 4.3.2 Over Windows 35 4.4 代码练习（以分组滚动窗口为例） 36 第五章 函数（Functions） 38 5.1 系统内置函数 38 5.2 UDF 40 5.2.1 注册用户自定义函数 UDF 40 5.2.2 标量函数（Scalar Functions） 40 5.2.3 表函数（Table Functions） 42 5.2.4 聚合函数（Aggregate Functions） 45 5.2.5 表聚合函数（Table Aggregate Functions） 47
StreamTableEnvironment在大数据中的应用及源码解析
PixelInk的博客
08-12 381
作为大数据处理框架Flink的一部分，Env-StreamTableEnvironment（以下简称StreamTableEnvironment）提供了一种方便且高效的流式SQL编程方式，使得开发人员能够以类似于关系型数据库的思维模式来处理和分析流式数据。它不仅可以处理实时流数据，还可以处理批处理数据。流式ETL：StreamTableEnvironment提供了丰富的数据转换和处理函数，可以对流式数据进行ETL（Extract-Transform-Load）操作，实现数据清洗、数据过滤、字段映射等功能。
Flink-Table SQL 实践编程 StreamTableEnvironment（十）下
springk的专栏
03-25 3351
StreamTableEnvironment主要是流式数据处理，本文主要实现GroupBy Window Aggregation、Over Window aggregation对应sql api的代码实现，同时实现流数据表join 连接。 代码中使用socket方式作为source数据输入方式，闲话不说，具体代码如下：StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
Flink-Table StreamTableEnvironment实践编程（四）下
springk的专栏
01-17 2315
七、实现UpsertStreamTableSink操作 public class TableStreamFlinkStudentUpsertTest { public static void main(String[] args) throws Exception { StreamExecutionEnvironment env = Stream...
Flink-Table StreamTableEnvironment实践编程（四）上
springk的专栏
01-17 5986
在上一篇文章中掌握了StreamTableEnvironment的基本概念，下面就进入代码实战环节： 一、实例中采用socket连接方式，获取数据流信息，使用 nc -l 端口号方式，开启端口监控，用来数据数据信息，实例中使用9999作为端口，命令行中输入：nc -l 9999 二、本实例主要功能是完成根据学生姓名实现分数统计 三、具体依赖实体类如下： @Data public ...
流环境Table API的使用基础
zg_hover的专栏
11-12 682
Flink的Table API在批和流环境下都有对应的类可以使用，本文主要介绍流环境（StreamExecutionEnvironment）下的Table API的使用。 本文只是一个使用的入门，所以会在本地单机环境下进行操作。 表的基本概念 临时表和永久表 表可以是临时的，并与单个 Flink 会话（session）的生命周期相关，也可以是永久的，并且在多个 Flink 会话和群集（cluster）中可见。 永久表需要 catalog（例如 Hive Metastore）以维护表的元数据。一旦永久表被创建
Flink流批一体计算（11）：PyFlink Tabel API之TableEnvironment
victory0508的专栏
07-05 1548
编写 Flink Python Table API 程序的第一步是创建 TableEnvironment。这是 Python Table API 作业的入口类。get_config()返回 table config，可以通过 table config 来定义 Table API 的运行时行为。
Flink Java Table API & SQL 之环境创建，表创建， 表查询，表输出
weixin_46376562的博客
07-06 1717
入门操作
Env-StreamTableEnvironment源码解析
VavzNvml的博客
09-19 180
通过注册数据源、执行SQL查询和输出结果，我们可以完成对流数据的处理任务。以上是对Env-StreamTableEnvironment源码的详细解析，通过阅读源码并结合示例代码，我们可以更好地理解和使用StreamTableEnvironment。在上述示例中，我们将结果DataStream对象添加了一个SinkFunction，你可以在invoke()方法中对结果数据进行处理。在上述示例中，我们创建了一个包含名字和年龄的DataStream对象，并将其注册为名为"people"的表。
flink批流一体编程实践
12-10
Flink 批流一体编程实践可以通过 Table/SQL API 和 DataStream/DataSet API 两种方式来实现。下面将分别介绍这两种方式的实现方法。 1. Table/SQL API 实现批流一体编程实践 Table/SQL API 是 Flink 提供的一种...
评论
添加红包

请填写红包祝福语或标题

个

红包个数最小为10个

元

红包金额最低5元

当前余额3.43元 前往充值 >
需支付：10.00元
成就一亿技术人!
领取后你会自动成为博主和红包主的粉丝 规则
hope_wisdom
发出的红包

打赏作者

springk

你的鼓励将是我创作的最大动力

¥1 ¥2 ¥4 ¥6 ¥10 ¥20
扫码支付：¥1
获取中
扫码支付

您的余额不足，请更换扫码支付或充值

打赏作者

实付元
使用余额支付
点击重新获取
扫码支付
钱包余额 0

抵扣说明：

1.余额是钱包充值的虚拟货币，按照1:1的比例进行支付金额的抵扣。
2.余额无法直接购买下载，可以购买VIP、付费专栏及课程。

余额充值