【Apache Flink】基于时间和窗口的算子-配置时间特性

这篇具有很好参考价值的文章主要介绍了【Apache Flink】基于时间和窗口的算子-配置时间特性。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

前言

【Apache Flink】基于时间和窗口的算子-配置时间特性,Apache Flink,apache,flink,大数据
Apache Flink 它提供了多种类型的时间和窗口概念,使得用户能够进行准确的时间计算。在数据处理任务中,时间的概念是非常重要的,对于一些复杂的实时流处理任务,如事件按时间顺序的聚合、分割和窗口计算,时间更是关键所在。而在这类任务中,选择使用何种时间特性是决定结果准确性的非常重要的一部分。Flink提供了三种时间特性供用户选择:事件时间、处理时间和摄取时间。

在使用Flink进行流处理时,时间窗口的选择也至关重要。Flink的窗口操作可以帮助我们将无限的流数据切分为有限的块,方便我们对每个数据块进行计算。Flink提供了多种窗口类型,包括滑动窗口、滚动窗口、会话窗口和全局窗口,用户可以根据业务需求选择合适的窗口类型。

本文将详细介绍Apache Flink中基于时间和窗口的算子,以及如何配置数据流的时间特性, 深入地理解和使用Flink在实际流式处理任务中进行数据计算的能力。

配置时间特性

在Apache Flink中,时间的概念是极其关键的要素,尤其是当我们处理实时或近实时数据流的时候。Flink提供了三种不同的时间概念,分别用于处理不同的任务和场景。

  1. 事件时间 (Event Time): 事件时间是指数据产生的时间,这个时间通常由事件数据本身携带,例如每个事件的日志行通常都会记录时间戳。在处理可能存在乱序或延迟数据的流计算任务时,事件时间是最常用的处理策略,因为它可以按照事件的发生顺序处理信息,而不是它们被系统接收的顺序。

  2. 处理时间 (Processing Time): 处理时间则是事件在系统中被处理时的系统时间,即事件在引擎处理时的“现在”时间。处理时间对于想要最大性能、毫秒级结果的场景非常适用,比如实时监控或者实时报警这样对处理速度有极高要求的场景。

  3. 摄取时间 (Ingestion Time): 摄取时间属于事件时间和处理时间的一种折衷方案。它是指事件进入Flink系统的时间。如果在一些场合,事件的时间戳无法获取或者不准确,同时又需要一定的事件排序,那么摄取时间就派上用场了。其在源头就分配了时间戳,并在整个处理过程中保留。摄取时间比处理时间语义强,比事件时间性能好。

这三种时间概念的选用,在很大程度上,决定了Flink处理事件的顺序和方式,因此根据实际的场景选择最适合的时间策略是非常重要的。

  1. 事件时间 (Event Time)
  env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

Flink会根据事件所携带的时间戳来处理数据,这就说明了数据是基于何时发生的进行处理的,而不是基于何时被处理的。比如在处理日志分析时,如果日志已经按照事件的发生时间排序,那么事件时间这种设置就会非常有用。

  1. 处理时间 (Processing Time)
   env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);

Flink会按照数据进入系统的时间即系统处理的实时进行处理,无视事件自身的时间戳。比如对于实时监控或者实时报警这样对处理速度有极高要求的场景,处理时间是最适合的。

  1. 摄取时间 (Ingestion Time)
  env.setStreamTimeCharacteristic(TimeCharacteristic.IngestionTime);

在数据读入Flink时,会自动地获取数据的摄取时间,与处理时间的观念类似但在数据进入系统时就已经赋予了时间。比如在希望事件能在一定程度上按照顺序处理,但又无法获取准确的事件时间时,摄取时间是一个不错的选择。

将时间特性设置为事件时间

在 Flink 中,我们可以指定时间特性为事件时间(Event Time),这是处理一些具体问题(例如延迟数据、重新处理等)的关键。下面是如何在 Flink 中设置时间特性为事件时间的代码示例:
读入一些元素,为它们分配时间戳和水印,并打印出来。分配时间戳和水印是在使用事件时间进行窗口操作等处理时必须要做的。

创建一个执行环境,并设置为事件时间。然后创建一个简单的字符串流,并分配时间戳和水印。最后我们使用了一个时间窗口并打印出了所有的元素。。

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;

public class EventTimeExample {
    public static void main(String[] args) throws Exception {
        // 创建执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        
        // 设置时间特性为事件时间
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);
        
        // 创建一个数据源,这里我们用的是从集合中获取
        DataStream<String> stream = env.fromElements("element1", "element2", "element3");
        
        // 分配时间戳和水印,这里我们假设通过某种方法获取了一个递增的时间戳
        DataStream<String> withTimestampsAndWatermarks = stream.assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
            @Override
            public long extractAscendingTimestamp(String element) {
                // 返回当前元素的时间戳,这儿仅作示例,实际应用需要根据具体需求获取
                return System.currentTimeMillis();
            }
        });

        // 对数据进行一些操作,比如过滤,这里的窗口大小为5秒
        withTimestampsAndWatermarks
          .timeWindowAll(Time.seconds(5))
          .apply(new AllWindowFunction<String, Object, TimeWindow>() {
              // 在应用函数中,我们做一些处理,这里简单地打印出所有元素
              public void apply(TimeWindow window, Iterable<String> values, Collector<Object> out) {
                  for (String value : values) {
                      System.out.println(value);
                  }
              }
          });

        // 启动应用
        env.execute("Event Time Example");
    }
}

时间戳分配器

在数据处理系统中,特别是在事件驱动的系统或实时流处理系统中,时间戳分配器是一个非常重要的组件。时间戳分配器的任务就是为每一个事件或数据记录分配一个时间戳,这个时间戳表示该事件发生的时间。

在 Apache Flink 这样的流处理框架中,时间戳分配器通常在数据源(Source)接收到数据时运行,为每一个事件分配一个时间戳。基于事件时间处理的窗口运算、Watermark 生成等流处理任务依赖于这个时间戳。

提供一个基础的时间戳分配器示例,此处假设有一个MapFunction,它将输入的MyEvent转换为一个包含时间戳和事件数据的Tuple2

MyTimestampAssigner实现了AssignerWithPunctuatedWatermarks接口,extractTimestamp()方法用于从数据元素中抽取出时间戳,checkAndGetNextWatermark()方法用于生成水印。在这里每处理一个元素都会调用checkAndGetNextWatermark()生成一个新的水印,这被称为 “punctuated” 模式。若要使用更常见的 “periodic” 模式,需要实现AssignerWithPeriodicWatermarks接口,这会定期(而不是每处理一个元素)生成水印。

public class MyTimestampAssigner implements AssignerWithPunctuatedWatermarks<MyEvent> {

    @Nullable
    @Override
    public Watermark checkAndGetNextWatermark(MyEvent event, long extractTimestamp) {
        return new Watermark(extractTimestamp);
    }

    @Override
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        return element.getCreationTime();
    }
}

然后,可以在流处理中使用这个MyTimestampAssigner来为数据流中的每一个元素分配时间戳:

DataStream<MyEvent> stream = ...
stream.assignTimestampsAndWatermarks(new MyTimestampAssigner());

源数据通常会有一个时间字段,它代表了数据的生成时间。我们可以基于这个时间字段来处理数据,并生成结果。为了让Flink知道每条数据的时间,我们需要自定义Timestamps/Watermarks。以下是一个简单的示例,该代码是在Flink中进行窗口计数的常见工作模式。

在水印生成部分设置了10秒的延迟时间,以处理乱序数据。这意味着,当窗口看到最新的水印时间大于其结束时间时,窗口才会触发进行执行计算。触发后的10秒内,该窗口还会接收更晚到达的数据。这对于处理乱序数据非常有用。

import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction.SourceContext;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.api.java.tuple.Tuple2;

public class TimeStampWaterMarkExample {
    public static void main(String[] args) throws Exception {
        // 创建执行环境,设置为事件时间模式
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 元组的第一个参数是一个字符串,第二个参数是一个时间戳
        Tuple2[] data = new Tuple2[]{
                new Tuple2<>("a", 1558430842000L),
                new Tuple2<>("b", 1558430843000L),
                new Tuple2<>("c", 1558430845000L)
        };

        // 添加数据源
        DataStreamSource<Tuple2<String, Long>> dataSource = env.addSource(new SourceFunction<Tuple2<String, Long>>() {
            @Override
            public void run(SourceContext<Tuple2<String, Long>> ctx) throws Exception {
                for (Tuple2<String, Long> datum : data) {
                    ctx.collectWithTimestamp(datum, datum.f1);
                    ctx.emitWatermark(new Watermark(datum.f1 - 1));
                }
                ctx.emitWatermark(new Watermark(Long.MAX_VALUE));
            }

            @Override
            public void cancel() {

            }
        });

        // 添加我们的时间戳分配器和水印生成器
        SingleOutputStreamOperator<Tuple2<String, Long>> timestampOperator = dataSource.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<Tuple2<String, Long>>(Time.seconds(10)) {
            @Override
            public long extractTimestamp(Tuple2<String, Long> element) {
                return element.f1;
            }
        });

        // 执行滚动窗口操作
        timestampOperator.keyBy(0)
                .window(TumblingEventTimeWindows.of(Time.seconds(2)))  // 定义滚动窗口
                .apply(new CountWindowFunction())
                .print();

        env.execute("job name");
    }
}

周期性水位线分配器

在流处理系统中,Watermark(水位线)是一种用于处理事件时间乱序的机制。在实时流处理系统中,来自不同源的事件可能以不同的顺序到达。这种顺序的不确定性会给处理系统带来挑战,因为系统必须确定何时所有相关事件都已到达,以便可以进行处理(例如,计算一个5分钟滑动窗口的平均值)。

Watermark就是流处理系统用于处理这种不确定性的一种手段。在Flink中,Watermark是一个特殊的事件或者信号,表示在此时间戳之前的所有事件都已经到达,没有更早的事件会到达。当Watermark t 到达时,说明在时间 t 或者更早的所有数据都已经收到,可以开始处理时间小于或等于 t 的窗口。

周期性水位线 (Periodic Watermark) 是一种定时生成的 Watermark。具体来说,数据流会周期性地调用 getCurrentWatermark() 方法获取新的 Watermark,并且插入到流中。这个周期默认是200ms,也可以通过 ExecutionConfig.setAutoWatermarkInterval(…) 来设置。

在定义 Watermark 生成逻辑时,通常会设置一个允许乱序到达的事件的最大延迟时间。例如,如果最大延迟时间设置为5秒,那么就意味着,Watermark t 只能保证时间戳小于 t-5s 的所有事件都已经到达。换句话说,水位线是滞后于事件时间的,即使水位线是周期性地生成的。

创建一个实现AssignerWithPeriodicWatermarks接口的类,目的是为了周期性生成watermark

BoundedOutOfOrdernessGenerator实现了AssignerWithPeriodicWatermarks接口。extractTimestamp()方法提取了每个元素的时间戳,getCurrentWatermark()则返回新的 watermark。这个实现假设元素可以最多晚3.5秒到达。使用当前最大时间戳减去这个延迟就得到了新的 watermark。这就意味着,即使有延迟的元素到达,只要其时间戳比当前的 watermark 大,它仍然可以用于窗口计算。

 
public class BoundedOutOfOrdernessGenerator implements AssignerWithPeriodicWatermarks<MyEvent> {

    // 最大的乱序容忍度设置为3500毫秒,也就是3.5秒
    private final long maxOutOfOrderness = 3500; 

    // 当前收到的最大的时间戳
    private long currentMaxTimestamp;

    @Override
    // 此方法用于从数据元素中提取时间戳
    public long extractTimestamp(MyEvent element, long previousElementTimestamp) {
        long timestamp = element.getCreationTime();
        // 更新当前收到的最大的时间戳
        currentMaxTimestamp = Math.max(timestamp, currentMaxTimestamp);
        // 返回提取的时间戳
        return timestamp;
    }

    @Nullable
    @Override
    // 此方法返回当前的Watermark,Watermark值为收到的最大时间戳减去乱序容忍度
    public Watermark getCurrentWatermark() {
        return new Watermark(currentMaxTimestamp - maxOutOfOrderness);
    }
}

在流处理中用这个BoundedOutOfOrdernessGenerator为流中的每个元素分配时间戳,并定期生成水位线。

DataStream<MyEvent> stream = ...
// 对流中的元素调用assignTimestampsAndWatermarks函数,传入的参数为BoundedOutOfOrdernessGenerator的实例,这样就完成了对流中元素的时间戳分配和周期性的Watermark生成
stream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessGenerator());

定点水位线分配器

定点水位线分配器(Fixed Watermark Assigner)是一种特定类型的水位线分配器,它生成一个固定的水位线值。它通常在事件的时间戳已经有序且无乱序出现时使用。由于每个元素都有一个时间戳,在这种情况下,水位线可能仅被推进到元素的当前时间戳。在一些特定的应用场景中,这样可能就够用了。

在Apache Flink中,可以使用assignTimestampsAndWatermarks函数结合WatermarkStrategy实现水位线的分配。对于定点水位线分配器,可以创建一个固定的,不变的水位线,例如:

WatermarkStrategy
    .<YourEvent>forMonotonousTimestamps()
    .withTimestampAssigner((event, timestamp) -> event.getTimestamp());

在这个例子中,forMonotonousTimestamps方法会创建一个定点水位线分配器,它将水位线固定在最近被处理的事件的时间戳。withTimestampAssigner方法定义了如何从事件中抽取时间戳。

定点水位线分配器用在那些时间戳严格递增的事件流中,如果在这样的流上使用定点水位线分配器,如果事件到达的顺序与时间戳的顺序不一致(例如,由于网络延迟、机器间的时钟偏移等),就有可能会得到错误的结果。

示例

在这个示例中假定事件是按照时间戳顺序处理的,所以可以有效地使用定点水位线分配器。如果事件的顺序可能会被打乱,那建议使用其他类型的水位线分配器。

创建一个水位线函数,对每一个输入的元素都分配它的时间戳,并产生连续定点的水位线。

public static class MyEvent {
    public long timestamp;  // 定义保存时间戳的变量
    public String data;  // 定义保存数据的变量

    public MyEvent(long timestamp, String data) {  // 构造函数
        this.timestamp = timestamp;
        this.data = data;
    }

    public long getTimestamp() {  // 获取时间戳的接口
        return timestamp;
    }
}

定义了事件类,包括时间戳和数据。文章来源地址https://www.toymoban.com/news/detail-715270.html

import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;

// 初始化执行环境
StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

// 配置事件时间特性
env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

// 输入的数据流
DataStream<MyEvent> inputStream = env.fromElements(
        new MyEvent(System.currentTimeMillis(), "data1"),
        new MyEvent(System.currentTimeMillis() + 1000, "data2"),
        new MyEvent(System.currentTimeMillis() + 2000, "data3")
);

// 使用水位线分配器给事件分配时间戳和水位线
DataStream<MyEvent> withTimestampsAndWatermarks = inputStream
        .assignTimestampsAndWatermarks(
                WatermarkStrategy.<MyEvent>forMonotonousTimestamps()
                        .withTimestampAssigner((event, timestamp) -> event.getTimestamp())
        );

参考文档

  1. https://nightlies.apache.org/flink/flink-docs-release-1.18/zh/

到了这里,关于【Apache Flink】基于时间和窗口的算子-配置时间特性的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

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

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

相关文章

  • 基于apache的http文件服务配置

    背景: 公司的产品使用的第三方模组可以OTA,厂家提供的是window开启软件,这样就可以在本机做http下载服务器,然后使用端口映射的方式,公开到外网,这样就可以进行4G网络访问内网服务器了。但这个有个弊端,需要电脑一直开着才有文件下载服务,端口也不是固定的,临

    2024年01月22日
    浏览(43)
  • Apache服务的搭建与配置,虚拟主机(基于域名 基于端口 基于Ip)

     1.目前最主流的三个Web服务器是Apache、Nginx、 IIS。  2. WEB 服务应用层使用HTTP协议。 3.web服务器:apache nginx IIS  #端口全部为80!https为443端口 4.Apache是世界使用排名第一的Web服务器软件 5.Apache的主程序名叫httpd。 http的默认发布网站的页面是 : /var/www/html 同时这个默认也可以改

    2024年02月03日
    浏览(47)
  • Apache Flink 和 Apache Kafka 两者之间的集成架构 Flink and Apache Kafka: A Winning Partnership

    作者:禅与计算机程序设计艺术 Apache Flink 和 Apache Kafka 是构建可靠、高吞吐量和低延迟的数据管道(data pipeline)的两个著名的开源项目。2019年4月,两者宣布合作共赢。在这次合作中,Apache Kafka 将提供强大的消息存储能力、Flink 将作为一个分布式数据流处理平台来对其进行

    2024年02月11日
    浏览(50)
  • Apache Doris 2.0.0 特性分析

    1、存算分离 所谓存算分离是指查询外表时,使用一种专门做计算的BE节点,但对于存储在BE上的内部表,目前还不能做到存储分离。 doris可以查询外部表,包括: Hive、Iceberg、Hudi、Elasticsearch、JDBC、Paimon 早期版本中,FE通过BE节点查询外部表,并且在BE节点内执行join、sort、

    2024年02月12日
    浏览(44)
  • 数据架构的实时分析:Apache Flink 和 Apache Storm 的比较

    实时数据处理在大数据领域具有重要意义,它可以帮助企业更快地获取和分析数据,从而更快地做出决策。随着数据量的增加,传统的批处理方法已经不能满足企业的需求,因此需要使用实时数据处理技术。 Apache Flink 和 Apache Storm 是两个流行的实时数据处理框架,它们都可以

    2024年01月23日
    浏览(53)
  • Apache Flink

    1 Flink的特点   1. 支持Scala和Java API。   2. 支持批流一体。   3. 同时支持高吞吐、低延迟、高性能。   4. 支持事件时间和处理时间语义,基于事件时间语义能够针对无序事件提供精确、一致的结果;基于处理时间语义能够用在具有极低延迟需求中。   5. 支持不同时

    2024年02月01日
    浏览(25)
  • 1.什么是Apache Flink?

    Flink专栏目录(点击进入…) 每个人的人生都只有一次机会,不要用最宝贵的青春为别人而活。 愿在试错中不断成长。 快来加入我们的互动学习群吧,与我们一起成长和进步吧! QQ交流群: 948912943 Apache Flink是一个框架和 分布式处理引擎 ,用于对无界和有界数据流进行状态

    2024年04月14日
    浏览(22)
  • 【跟小嘉学 Apache Flink】二、Flink 快速上手

    【跟小嘉学 Apache Flink】一、Apache Flink 介绍 【跟小嘉学 Apache Flink】二、Flink 快速上手 创建 maven 工程 并且添加如下依赖 在 resource 目录下创建 log4j.properties 文件,写入如下内容 实际上在 Flink 里面已经做到流批处理统一,官方推荐使用 DateStream API,在跳任务时通过执行模式设

    2024年02月09日
    浏览(35)
  • Apache Flink从入门到放弃——Flink简介(一)

       随着大数据的发展,大数据的存储、计算、运用百花齐放;而大数据的计算中最重要的就是计算引擎,时至今日,很多人将大数据引擎分为四代,分别是: 第一代,Hadoop承载的MapReduce,将计算分为Map和Reduce两个阶段,同时采用Hadoop集群的分布式计算原理来实现数据的计

    2024年02月05日
    浏览(38)
  • 【Apache Flink】实现有状态函数

    Flink为键值分区状态(Keyed State)提供了几种不同的原语(数据类型)。这是因为不同的算法和操作可能需要管理不同类型的状态。其中一些原语包括: ValueState : 这种状态类型用于存储单个的,可能更新的值。常见的用途包括存储计数器或聚合。 ListState : 这种状态用于存储一

    2024年02月08日
    浏览(35)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包