Flink 算子:数据处理的魔法师

这篇具有很好参考价值的文章主要介绍了Flink 算子:数据处理的魔法师。希望对大家有所帮助。如果存在错误或未考虑完全的地方,请大家不吝赐教,您也可以点击"举报违法"按钮提交疑问。

导语

用户通过算子能将一个或多个 DataStream 转换成新的 DataStream,在应用程序中可以将多个数据转换算子合并成一个复杂的数据流拓扑。

这部分内容将描述 Flink DataStream API 中基本的数据转换 API,数据转换后各种数据分区方式,以及算子的链接策略。

数据流转换

1.Map输入一个元素同时输出一个元素

java:

	 DataStream<Integer> dataStream = //...

	dataStream.map(new MapFunction<Integer, Integer>() {

	  @Override

  	 	public Integer map(Integer value) throws Exception {

	       return 2 * value;

	    }

	});

scala:

dataStream.map { x => x * 2 }

2.FlatMap输入一个元素同时产生零个、一个或多个元素。

java:

dataStream.flatMap(new FlatMapFunction<String, String>() {

    @Override

  public void flatMap(String value, Collector<String> out) throws Exception {

		  for(String word: value.split(" ")){

            out.collect(word);

        }

    }

});

scala:

 dataStream.flatMap { str => str.split(" ") }

3.Filter为每个元素执行一个布尔 function,并保留那些 function 输出值为 true 的元素

java:

true 的元素。</font>
	dataStream.filter(new FilterFunction<Integer>() {
    @Override
    public boolean filter(Integer value) throws Exception {
        return value != 0;
    }
});

sclala:

	dataStream.filter { _ != 0 }

4.KeyBy在逻辑上将流划分为不相交的分区。具有相同 key 的记录都分配到同一个分区。在内部, keyBy() 是通过哈希分区实现的。有多种指定 key的方式。
java:

	dataStream.keyBy(value -> value.getSomeKey());
	dataStream.keyBy(value -> value.f0);

scala:

	dataStream.keyBy(_.someKey)
	dataStream.keyBy(_._1)

以下情况,一个类不能作为 key

1.它是一种 POJO 类,但没有重写 hashCode() 方法而是依赖于 Object.hashCode() 实现。
2. 它是任意类的数组。

5.Reduce:在相同 key 的数据流上“滚动”执行 reduce。将当前元素与最后一次 reduce 得到的值组合然后输出新值。
java:

	keyedStream.reduce(new ReduceFunction<Integer>() {
    @Override
    public Integer reduce(Integer value1, Integer value2)  throws Exception {
        return value1 + value2;
    }
});

scala:

	keyedStream.reduce { _ + _ }

6.Window:可以在已经分区的 KeyedStreams 上定义 Window。Window 根据某些特征(例如,最近 5 秒内到达的数据)对每个 key Stream 中的数据进行分组。
java:

dataStream
  .keyBy(value -> value.f0)
  .window(TumblingEventTimeWindows.of(Time.seconds(5)));

scala:

dataStream
  .keyBy(_._1)
  .window(TumblingEventTimeWindows.of(Time.seconds(5)))

7.WindowAll:可以在普通 DataStream 上定义 Window。 Window 根据某些特征(例如,最近 5 秒内到达的数据)对所有流事件进行分组。

java:

	dataStream
	  .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)));

scala:

dataStream
  .windowAll(TumblingEventTimeWindows.of(Time.seconds(5)))

8.Window Apply:将通用 function 应用于整个窗口

如果你使用 windowAll 转换,则需要改用 AllWindowFunction

java:

windowedStream.apply(new WindowFunction<Tuple2<String,Integer>, Integer, Tuple, Window>() {
    public void apply (Tuple tuple,
            Window window,
            Iterable<Tuple2<String, Integer>> values,
            Collector<Integer> out) throws Exception {
        int sum = 0;
        for (value t: values) {
            sum += t.f1;
        }
        out.collect (new Integer(sum));
    }
});
// 在 non-keyed 窗口流上应用 AllWindowFunction
allWindowedStream.apply (new AllWindowFunction<Tuple2<String,Integer>, Integer, Window>() {
    public void apply (Window window,
            Iterable<Tuple2<String, Integer>> values,Collector<Integer> out) throws Exception {
        int sum = 0;
        for (value t: values) {
            sum += t.f1;
        }
        out.collect (new Integer(sum));
    }
});

scala:

	windowedStream.apply { WindowFunction }
	// 在 non-keyed 窗口流上应用 AllWindowFunction
	allWindowedStream.apply { AllWindowFunction }

9.WindowReduce:对窗口应用 reduce function 并返回 reduce 后的值

java:

windowedStream.reduce (new ReduceFunction<Tuple2<String,Integer>>() {
    public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1,Tuple2<String, Integer> value2) throws Exception {
        return new Tuple2<String,Integer>(value1.f0, value1.f1 + value2.f1);
    }
});

scala:

	windowedStream.reduce { _ + _ }

10.Union:将两个或多个数据流联合来创建一个包含所有流中数据的新流。注意:如果一个数据流和自身进行联合,这个流中的每个数据将在合并后的流中出现两次。

java:

dataStream.union(otherStream1, otherStream2, ...);

scala:

dataStream.union(otherStream1, otherStream2, ...)

11.Window Join根据指定的 key 和窗口 join 两个数据流。

java:

dataStream.join(otherStream)
    .where(<key selector>).equalTo(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new JoinFunction () {...});

scala:

dataStream.join(otherStream)
    .where(<key selector>).equalTo(<key selector>)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply { ... }

12.Interval Join根据 key 相等并且满足指定的时间范围内的条件将分别属于两个 keyed stream 的元素 e1 和 e2 Join 在一起。

java:

// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2

keyedStream.intervalJoin(otherKeyedStream)
    .between(Time.milliseconds(-2), Time.milliseconds(2))      // lower and upper bound
    .upperBoundExclusive(true)      // optional
    .lowerBoundExclusive(true)       // optional
    .process(new IntervalJoinFunction() {...});

scala:

// this will join the two streams so that
// key1 == key2 && leftTs - 2 < rightTs < leftTs + 2

keyedStream.intervalJoin(otherKeyedStream)
    .between(Time.milliseconds(-2), Time.milliseconds(2))        // lower and upper bound
    .upperBoundExclusive(true)     // optional
    .lowerBoundExclusive(true)      // optional
    .process(new IntervalJoinFunction() {...})

13.Window CoGroup:根据指定的 key 和窗口将两个数据流组合在一起。

java:

dataStream.coGroup(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply (new CoGroupFunction () {...});

scala:

dataStream.coGroup(otherStream)
    .where(0).equalTo(1)
    .window(TumblingEventTimeWindows.of(Time.seconds(3)))
    .apply {}

14.Connect:“连接” 两个数据流并保留各自的类型。connect 允许在两个流的处理逻辑之间共享状态。

java:

DataStream<Integer> someStream = //...
DataStream<String> otherStream = //...
ConnectedStreams<Integer, String> connectedStreams = someStream.connect(otherStream);

scala:

someStream : DataStream[Int] = ...
otherStream : DataStream[String] = ...
val connectedStreams = someStream.connect(otherStream)

15.CoMap, CoFlatMap:类似于在连接的数据流上进行 map 和 flatMap。

java:

connectedStreams.map(new CoMapFunction<Integer, String, Boolean>() {
    @Override
    public Boolean map1(Integer value) {
        return true;
    }
    @Override
    public Boolean map2(String value) {
        return false;
    }
});
connectedStreams.flatMap(new CoFlatMapFunction<Integer, String, String>() {
   @Override
   public void flatMap1(Integer value, Collector<String> out) {
       out.collect(value.toString());
   }
   @Override
   public void flatMap2(String value, Collector<String> out) {
       for (String word: value.split(" ")) {
         out.collect(word);
       }
   }
});

scala:

connectedStreams.map(
    (_ : Int) => true,
    (_ : String) => false
)
connectedStreams.flatMap(
    (_ : Int) => true,
    (_ : String) => false
)

16.Iterate:通过将一个算子的输出重定向到某个之前的算子来在流中创建“反馈”循环。这对于定义持续更新模型的算法特别有用。

//下面的代码从一个流开始,并不断地应用迭代自身。大于 0 的元素被发送回反馈通道,其余元素被转发到下游。
java:
IterativeStream<Long> iteration = initialStream.iterate();
DataStream<Long> iterationBody = iteration.map (/*do something*/);
DataStream<Long> feedback = iterationBody.filter(new FilterFunction<Long>(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value > 0;
    }
});
iteration.closeWith(feedback);
DataStream<Long> output = iterationBody.filter(new FilterFunction<Long>(){
    @Override
    public boolean filter(Long value) throws Exception {
        return value <= 0;
    }
});

scala

//下面的代码从一个流开始,并不断地应用迭代自身。大于 0 的元素被发送回反馈通道,其余元素被转发到下游。
initialStream.iterate {
  iteration => {
    val iterationBody = iteration.map {/*do something*/}
    (iterationBody.filter(_ > 0), iterationBody.filter(_ <= 0))
  }
}

物理分区

1.自定义分区:使用用户定义的 Partitioner 为每个元素选择目标任务。

Java

dataStream.partitionCustom(partitioner, "someKey");
dataStream.partitionCustom(partitioner, 0);

Scala

dataStream.partitionCustom(partitioner, "someKey")
dataStream.partitionCustom(partitioner, 0)

2.随机分区:将元素随机地均匀划分到分区。

Java

dataStream.shuffle();

Scala

dataStream.shuffle()

2.Rescaling

将元素以 Round-robin轮询的方式分发到下游算子。
如果你想实现数据管道,这将很有用,例如,想将数据源多个并发实例的数据分发到多个下游 map来实现负载分配,但又不想像 rebalance() 那样引起完全重新平衡。该算子将只会到本地数据传输而不是网络数据传输,这取决于其它配置值,例如 TaskManager 的 slot 数量。

上游算子将元素发往哪些下游的算子实例集合同时取决于上游和下游算子的并行度。例如,如果上游算子并行度为 2,下游算子的并发度为 6,那么上游算子的其中一个并行实例将数据分发到下游算子的三个并行实例,另外一个上游算子的并行实例则将数据分发到下游算子的另外三个并行实例中。再如,当下游算子的并行度为2,而上游算子的并行度为 6 的时候,那么上游算子中的三个并行实例将会分发数据至下游算子的其中一个并行实例,而另外三个上游算子的并行实例则将数据分发至另下游算子的另外一个并行实例。

当算子的并行度不是彼此的倍数时,一个或多个下游算子将从上游算子获取到不同数量的输入。

请参阅下图来可视化地感知上述示例中的连接模式:
Flink 算子:数据处理的魔法师,Flink,flink

Java

dataStream.rescale();

Scala

dataStream.rescale()

广播:将元素广播到每个分区 。

Java

dataStream.broadcast();

Scala

dataStream.broadcast()

算子链和资源组

将两个算子链接在一起能使得它们在同一个线程中执行,从而提升性能。Flink 默认会将能链接的算子尽可能地进行链接(例如, 两个 map 转换操作)。此外, Flink 还提供了对链接更细粒度控制的 API 以满足更多需求:

如果想对整个作业禁用算子链,可以调用 StreamExecutionEnvironment.disableOperatorChaining()。下列方法还提供了更细粒度的控制。需要注意的是,这些方法只能在 DataStream 转换操作后才能被调用,因为它们只对前一次数据转换生效。例如,可以 someStream.map(...).startNewChain() 这样调用,而不能 someStream.startNewChain() 这样。

一个资源组对应着 Flink 中的一个 slot 槽,更多细节请看 slots 。 你可以根据需要手动地将各个算子隔离到不同的 slot 中。

创建新链

基于当前算子创建一个新的算子链。

后面两个 map 将被链接起来,而 filter 和第一个 map 不会链接在一起。

Java

someStream.filter(...).map(...).startNewChain().map(...);

Scala

someStream.filter(...).map(...).startNewChain().map(...)

禁止链接

禁止和 map 算子链接在一起。

Java

 someStream.map(...).disableChaining();

Scala

someStream.map(...).disableChaining()

配置 Slot 共享组

为某个算子设置 slot 共享组。Flink 会将同一个 slot 共享组的算子放在同一个 slot 中,而将不在同一 slot 共享组的算子保留在其它 slot 中。这可用于隔离 slot 。如果所有输入算子都属于同一个 slot 共享组,那么 slot 共享组从将继承输入算子所在的 slot。slot 共享组的默认名称是 “default”,可以调用 slotSharingGroup(“default”) 来显式地将算子放入该组。

Java

someStream.filter(...).slotSharingGroup("name");

Scala

someStream.filter(...).slotSharingGroup("name")

名字和描述

Flink里的算子和作业节点会有一个名字和一个描述。名字和描述。名字和描述都是用来介绍一个算子或者节点是在做什么操作,但是他们会被用在不同地方。

名字会用在用户界面、线程名、日志、指标等场景。节点的名字会根据节点中算子的名字来构建。 名字需要尽可能的简洁,避免对外部系统产生大的压力。

描述主要用在执行计划展示,以及用户界面展示。节点的描述同样是根据节点中算子的描述来构建。 描述可以包括详细的算子行为的信息,以便我们在运行时进行debug分析。

Java

someStream.filter(...).setName("filter").setDescription("x in (1, 2, 3, 4) and y > 1");

Scala

someStream.filter(...).setName("filter").setDescription("x in (1, 2, 3, 4) and y > 1")

节点的描述默认是按照一个多行的树形结构来构建的,用户可以通过把pipeline.vertex-description-mode设为CASCADING, 实现将描述改为老版本的单行递归模式。

Flink SQL框架生成的算子默认会有一个由算子的类型以及id构成的名字,以及一个带有详细信息的描述。 用户可以通过将table.optimizer.simplify-operator-name-enabled设为false,将名字改为和以前的版本一样的详细描述。

当一个作业的拓扑很复杂时,用户可以把pipeline.vertex-name-include-index-prefix设为true,在节点的名字前增加一个拓扑序的前缀,这样就可以很容易根据指标以及日志的信息快速找到拓扑图中对应节点。

Flink 1.15文章来源地址https://www.toymoban.com/news/detail-814496.html

到了这里,关于Flink 算子:数据处理的魔法师的文章就介绍完了。如果您还想了解更多内容,请在右上角搜索TOY模板网以前的文章或继续浏览下面的相关文章,希望大家以后多多支持TOY模板网!

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

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

相关文章

  • 【flink番外篇】2、flink的23种算子window join 和interval join 数据倾斜、分区介绍及详细示例(3)- 数据倾斜处理、分区示例

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

    2024年02月03日
    浏览(56)
  • Spark大数据处理学习笔记(3.2.2)掌握RDD算子

    衔接上文:http://t.csdn.cn/Z0Cfj 功能: reduce()算子按照传入的函数进行归约计算 案例: 计算1 + 2 + 3 + …+100的值 计算1 × 2 × 3 × 4 × 5 × 6 的值(阶乘 - 累乘) 计算1 2 + 2 2 + 3 2 + 4 2 + 5**2的值(先映射,后归约) 功能: collect()算子向Driver以数组形式返回数据集的所有元素。通常对

    2024年02月08日
    浏览(47)
  • Flink:处理大规模复杂数据集的最佳实践深入探究Flink的数据处理和性能优化技术

    作者:禅与计算机程序设计艺术 随着互联网、移动互联网、物联网等新型网络技术的不断发展,企业对海量数据的处理日益依赖,而大数据分析、决策支持、风险控制等领域都需要海量的数据处理能力。如何高效、快速地处理海量数据、提升处理效率、降低成本,是当下处理

    2024年02月13日
    浏览(56)
  • 【Flink基础】-- 延迟数据的处理

    目录 ​一、关于延迟的一些概念 1、什么是延迟? 2、什么导致互联网延迟?

    2024年02月03日
    浏览(43)
  • Flink对迟到数据的处理

    Apache Flink 是一个分布式流处理框架,可以通过多种方式处理延迟数据。这里有几个选项: 🍊水位线WaterMarker:Flink 使用水位线来跟踪流中的时间进度。水位线是由源发出的周期性时间戳,用于确定一条数据的延迟时间。您可以根据水位线指定数据的最大延迟。例如,您可以

    2024年02月06日
    浏览(34)
  • Flink实时大数据处理性能测试

    Flink是一个开源的流处理框架,用于实时大数据处理。它可以处理大量数据,提供低延迟和高吞吐量。Flink的性能测试是一项重要的任务,可以帮助我们了解其在实际应用中的表现。在本文中,我们将讨论Flink实时大数据处理性能测试的背景、核心概念、算法原理、代码实例、

    2024年03月18日
    浏览(50)
  • 从批处理到实时处理:Flink的数据处理变革和API扩展

    作者:禅与计算机程序设计艺术 Apache Flink是一个开源的分布式流处理平台,它由Apache Software Foundation(ASF)开发并于2015年9月发布。Apache Flink支持多种编程语言如Java、Scala、Python等进行编写,并且提供丰富的API接口方便用户进行数据处理。Flink的系统架构主要包括:JobManager、

    2024年02月12日
    浏览(43)
  • Flink流处理案例:实时数据聚合

    Apache Flink是一个流处理框架,可以处理大规模数据流,实现实时数据处理和分析。Flink支持各种数据源和接口,如Kafka、HDFS、TCP流等,可以实现高吞吐量、低延迟的流处理。 在本文中,我们将通过一个实际的Flink流处理案例来讲解Flink的核心概念、算法原理和最佳实践。我们将

    2024年02月19日
    浏览(45)
  • Flink之数据乱序处理

    在了解为什么会乱序之前我们先来看一下在Flink中的时间语序。 1.1  Flink Time  时间语义 Event Time:事件产生的时间,它通常由事件中的时间戳描述。 Ingestion Time:事件进入Flink的时间。 Processing Time:事件被处理时当前系统的时间。 这三种时间的对应关系如下图所示: 1.2  数

    2024年02月07日
    浏览(38)
  • 【大数据】Flink 架构(三):事件时间处理

    《 Flink 架构 》系列(已完结),共包含以下 6 篇文章: Flink 架构(一):系统架构 Flink 架构(二):数据传输 Flink 架构(三):事件时间处理 Flink 架构(四):状态管理 Flink 架构(五):检查点 Checkpoint(看完即懂) Flink 架构(六):保存点 Savepoint 😊 如果您觉得这篇

    2024年02月21日
    浏览(35)

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

支付宝扫一扫打赏

博客赞助

微信扫一扫打赏

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

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

二维码1

领取红包

二维码2

领红包