当前位置: 开发笔记 > 编程语言 > 正文

【大数据开发】Flink之DataSet转换操作2（九）

作者：小木马 | 来源：互联网 | 2023-09-09 14:31

点击蓝字关注我们Flink之DataSet转换操作2（九）01TransForm之Reduce操作通过两两合并将数据集中的元素合并成一个元素，可以在整

点击蓝字 关注我们

Flink之DataSet转换操作2（九）

TransForm之Reduce操作

通过两两合并将数据集中的元素合并成一个元素，可以在整个数据集上使用。

■将数据合并

val dataSet = env.fromElements(1,2,3,4)
val results = dataSet.reduce((x,y) => (x+y))

//Java
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class ReduceJavaDemo {
public static void main(String[] args) throws Exception{
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
);

// DataSet> result = source.flatMap(new FlatMapFunction>() {
// @Override
// public void flatMap(String line, Collector> collector) throws Exception {
// for(String word : line.toUpperCase().split(" ")){
// collector.collect(new Tuple2<>(word , 1));
// }
// }
// })
// .groupBy("f0")
// .reduce(new ReduceFunction>() {
// @Override
// public Tuple2 reduce(Tuple2 t1, Tuple2 t2) throws Exception {
// return new Tuple2<>(t1.f0,t1.f1+t2.f1);
// }
// });

DataSet<Tuple2<String,Integer>> result = source.flatMap((String line , Collector<Tuple2<String,Integer>> collector) -> {
for(String word : line.toUpperCase().split(" ")){
collector.collect(new Tuple2<>(word,1));
}
}).returns(Types.TUPLE(Types.STRING,Types.INT))
.groupBy("f0")
.reduce((x , y) -> new Tuple2<>(x.f0,x.f1+y.f1));
result.print();
}
}

//Scala
import org.apache.flink.api.scala.{ExecutionEnvironment, _}
import org.apache.flink.util.Collector

object ReduceScalaDemo {
def main(args: Array[String]): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment

val source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
)

val result = source
.flatMap((line : String,collector : Collector[(String,Int)]) => {
(line.toUpperCase.split(" ")).foreach(word => (collector.collect((word,1))))
})
.groupBy("_1")
.reduce((x,y) => (x._1,x._2+y._2))
.print()
}
}

TransForm之ReduceGroup操作

将一组元素合并成一个或者多个元素，可以在整个数据集上使用。这是对reduce程序的一个小优化。

■优化数据合并

val dataSet = env.fromElements(1,2,3,4)
val results = dataSet.reduceFroup(in => in reduce(x,y) => (x+y))

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class ReduceGroupJavaDemo {
public static void main(String[] args) throws Exception{
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
);

// DataSet> result = source
// .flatMap(new FlatMapFunction>() {
// @Override
// public void flatMap(String line, Collector> collector) throws Exception {
// for(String word : line.toUpperCase().split(" ")){
// collector.collect(new Tuple2<>(word,1));
// }
// }
// })
// .groupBy("f0")
// .reduceGroup(new GroupReduceFunction, Tuple2>() {
// @Override
// public void reduce(Iterable> iterable, Collector> collector) throws Exception {
// String word = null;
// int count = 0;
// for(Tuple2 tuple : iterable){
// word = tuple.f0;
// count += tuple.f1;
// }
// collector.collect(new Tuple2<>(word,count));
// }
// });

DataSetString,Integer>> result = source
.flatMap((String line ,CollectorString,Integer>> collector) -> {
for(String word : line.toUpperCase().split(" ")){
collector.collect(new Tuple2<>(word,1));
}
}).returns(Types.TUPLE(Types.STRING,Types.INT))
.groupBy("f0")
.reduceGroup((IterableString,Integer>> iterable , CollectorString,Integer>> collector) -> {
String word = null;
int count = 0;
for(Tuple2<String,Integer> tuple : iterable){
word = tuple.f0;
count += tuple.f1;
}
collector.collect(new Tuple2<>(word,count));
}).returns(Types.TUPLE(Types.STRING,Types.INT));
result.print();
}
}

TransForm之CombineGroup操作

我们可以通过CombineGroup事先在每一台机器上进行聚合操作，再通过ReduceGroup将每台机器CombineGroup输出的结果进行聚合,

这样的话，ReduceGroup需要汇总的数据量就少很多，从而加快计算的速度。

■优化数据合并

.groupBy("_1")
.combineGroup((words , out : Collector[(String , Int)]) =>{
out.collect(words reduce((x,y)=> (х. _1,x. _2+y. 2))
})
.groupBy("_1")
.reduceGroup(x => x.reduce((x,y)=> (x.1,x._2+y._2))

//Java
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.GroupCombineFunction;
import org.apache.flink.api.common.functions.GroupReduceFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class CombineGroupJavaDemo {
public static void main(String[] args) throws Exception{
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
DataSet<String> source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
);
// DataSet> result = source
// .flatMap(new FlatMapFunction>() {
// @Override
// public void flatMap(String line, Collector> collector) throws Exception {
// for(String word : line.toUpperCase().split(" ")){
// collector.collect(new Tuple2<>(word,1));
// }
// }
// })
//
// //本地上进行聚合操作
// .groupBy("f0")
// .combineGroup(new GroupCombineFunction, Tuple2>() {
// @Override
// public void combine(Iterable> iterable, Collector> collector) throws Exception {
// String word = null;
// int count =0;
// for(Tuple2 tuple : iterable){
// word = tuple.f0;
// count += tuple.f1;
// }
// collector.collect(new Tuple2<>(word,count));
// }
// })
// .groupBy("f0")
// .reduceGroup(new GroupReduceFunction, Tuple2>() {
// @Override
// public void reduce(Iterable> iterable, Collector> collector) throws Exception {
// String word = null;
// int count = 0;
// for(Tuple2 tuple : iterable){
// word = tuple.f0;
// count += tuple.f1;
// }
// collector.collect(new Tuple2<>(word,count));
// }
// });

DataSetString,Integer>> result = source
.flatMap((String line , CollectorString,Integer>> collector) -> {
for(String word : line.toUpperCase().split(" ")){
collector.collect(new Tuple2<>(word,1));
}
}).returns(Types.TUPLE(Types.STRING,Types.INT))
.groupBy("f0")
.combineGroup((IterableString,Integer>> iterable , CollectorString,Integer>> collector) -> {
String word = null;
int count = 0;
for(Tuple2<String,Integer> tuple : iterable){
word = tuple.f0;
count += tuple.f1;
}
collector.collect(new Tuple2<>(word,count));
}).returns(Types.TUPLE(Types.STRING,Types.INT))
.groupBy("f0")
.reduceGroup((IterableString,Integer>> iterable ,CollectorString,Integer>> collector) -> {
String word = null;
int count = 0 ;
for(Tuple2<String,Integer> tuple : iterable){
word =tuple.f0;
count += tuple.f1;
}
collector.collect(new Tuple2<>(word,count));
}).returns(Types.TUPLE(Types.STRING,Types.INT));

result.print();
}
}

//Scala
import org.apache.flink.api.scala.{ExecutionEnvironment,_}
import org.apache.flink.util.Collector

object CombineGroupScalaDemo {
def main(args: Array[String]): Unit = {
val env = ExecutionEnvironment.getExecutionEnvironment

val source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
)

val result = source
.flatMap((line : String , collector : Collector[(String,Int)]) => {
line.toUpperCase.split(" ").foreach(word => (collector.collect(word,1)))
})
.groupBy("_1")
.combineGroup((iterator , combine_collector : Collector[(String,Int)] )=> {
combine_collector.collect(iterator reduce((x,y) => (x._1,x._2+y._2)))
})
.groupBy("_1")
.reduceGroup((iterator , collector : Collector[(String,Int)]) => {
collector.collect(iterator reduce((x,y) => (x._1,x._2+y._2)))
}).print()
}
}

TransForm之Aggregate操作

通过Aggregate Function将一组元素值合并成单个值，可以在整个DataSet数据集上使用通过Aggregate Function将一组元素值合并成单个值，可以在整个DataSet数据集上使用。

■取合并后数值的最大值

.group("_1").aggregate(Aggregations.SUM,1)
.group("_1").aggregate(Aggregations.SUM,1).max(1)
.group("_1").aggregate(Aggregations.SUM,1).maxby(1)

//Java
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;

public class AggregateJavaDemo {
public static void main(String[] args) throws Exception{
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

DataSet<String> source = env.fromElements(
"spark hbase java",
"java spark hive",
"java hbase hbase"
);

DataSetString,Integer>> result = source
.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
@Override
public void flatMap(String s, CollectorString, Integer>> collector) throws Exception {
for(String word : s.toUpperCase().split(" ")){
collector.collect(new Tuple2<>(word,1));
}
}
})
.groupBy("f0")
.aggregate(Aggregations.SUM,1);
result.print();
}
}

扫描二维码

关注我们

微信号 : BIGDT_IN

推荐阅读

java
Java太阳系小游戏分析和源码详解

本文介绍了一个基于Java的太阳系小游戏的分析和源码详解。通过对面向对象的知识的学习和实践，作者实现了太阳系各行星绕太阳转的效果。文章详细介绍了游戏的设计思路和源码结构，包括工具类、常量、图片加载、面板等。通过这个小游戏的制作，读者可以巩固和应用所学的知识，如类的继承、方法的重载与重写、多态和封装等。 ... [详细]

蜡笔小新 2023-12-14 19:53:34
js
JS进修笔记——闭包的运转机制和作用域

本文介绍了闭包的定义和运转机制，重点解释了闭包如何能够接触外部函数的作用域中的变量。通过词法作用域的查找规则，闭包可以访问外部函数的作用域。同时还提到了闭包的作用和影响。 ... [详细]

蜡笔小新 2023-12-14 18:45:00
go
在类中定义数组时出错 - Error on defining arrays in class

Iamtryingtomakeaclassthatwillreadatextfileofnamesintoanarray,thenreturnthatarra ... [详细]

蜡笔小新 2023-12-14 17:38:12
function
实现下拉列表，点击其他位置自动隐藏效果的三种方式比较

目录实现效果：实现环境实现方法一：基本思路主要代码JavaScript代码总结方法二主要代码总结方法三基本思路主要代码JavaScriptHTML总结实 ... [详细]

蜡笔小新 2023-12-14 15:03:14
java
Java容器中的compareto方法排序原理解析

本文从源码解析Java容器中的compareto方法的排序原理，讲解了在使用数组存储数据时的限制以及存储效率的问题。同时提到了Redis的五大数据结构和list、set等知识点，回忆了作者大学时代的Java学习经历。文章以作者做的思维导图作为目录，展示了整个讲解过程。 ... [详细]

蜡笔小新 2023-12-14 13:53:31
java
JavaSE笔试题-接口、抽象类、多态等问题解答

本文解答了JavaSE笔试题中关于接口、抽象类、多态等问题。包括Math类的取整数方法、接口是否可继承、抽象类是否可实现接口、抽象类是否可继承具体类、抽象类中是否可以有静态main方法等问题。同时介绍了面向对象的特征，以及Java中实现多态的机制。 ... [详细]

蜡笔小新 2023-12-14 10:01:13
java
Spring特性实现接口多类的动态调用详解

本文详细介绍了如何使用Spring特性实现接口多类的动态调用。通过对Spring IoC容器的基础类BeanFactory和ApplicationContext的介绍，以及getBeansOfType方法的应用，解决了在实际工作中遇到的接口及多个实现类的问题。同时，文章还提到了SPI使用的不便之处，并介绍了借助ApplicationContext实现需求的方法。阅读本文，你将了解到Spring特性的实现原理和实际应用方式。 ... [详细]

蜡笔小新 2023-12-14 03:24:19
java
Android开发笔记：使用Picasso加载网络图片等比例缩放

在Android开发中，使用Picasso库可以实现对网络图片的等比例缩放。本文介绍了使用Picasso库进行图片缩放的方法，并提供了具体的代码实现。通过获取图片的宽高，计算目标宽度和高度，并创建新图实现等比例缩放。 ... [详细]

蜡笔小新 2023-12-14 17:34:00
java
向QTextEdit拖放文件的方法及实现步骤

本文介绍了在使用QTextEdit时如何实现拖放文件的功能，包括相关的方法和实现步骤。通过重写dragEnterEvent和dropEvent函数，并结合QMimeData和QUrl等类，可以轻松实现向QTextEdit拖放文件的功能。详细的代码实现和说明可以参考本文提供的示例代码。 ... [详细]

蜡笔小新 2023-12-14 16:06:38
java
Java猜拳小游戏代码

本文介绍了一个Java猜拳小游戏的代码，通过使用Scanner类获取用户输入的拳的数字，并随机生成计算机的拳，然后判断胜负。该游戏可以选择剪刀、石头、布三种拳，通过比较两者的拳来决定胜负。 ... [详细]

蜡笔小新 2023-12-14 15:39:08
function
Backwardsincompatible change made.

Commit1ced2a7433ea8937a1b260ea65d708f32ca7c95eintroduceda+Clonetraitboundtom ... [详细]

蜡笔小新 2023-12-14 15:35:09
js
开发笔记:加密&json&StringIO模块&BytesIO模块

篇首语：本文由编程笔记#小编为大家整理，主要介绍了加密&json&StringIO模块&BytesIO模块相关的知识，希望对你有一定的参考价值。一、加密加密 ... [详细]

蜡笔小新 2023-12-14 15:18:35
datetime
阿里云物联网 .NET Core 客户端 | CZGL.AliIoTClient：4. 设备上报属性

阿,里,云,物,联网,net,core,客户端,czgl,aliiotclient, ... [详细]

蜡笔小新 2023-12-14 12:40:20
js
OC学习笔记之@property和@synthesize

本文介绍了OC学习笔记中的@property和@synthesize，包括属性的定义和合成的使用方法。通过示例代码详细讲解了@property和@synthesize的作用和用法。 ... [详细]

蜡笔小新 2023-12-14 12:05:06
go
Open judge C16H: Magical Balls 快速幂+逆元问题解析

本文主要解析了Open judge C16H问题中涉及到的Magical Balls的快速幂和逆元算法，并给出了问题的解析和解决方法。详细介绍了问题的背景和规则，并给出了相应的算法解析和实现步骤。通过本文的解析，读者可以更好地理解和解决Open judge C16H问题中的Magical Balls部分。 ... [详细]

蜡笔小新 2023-12-14 12:03:27

小木马

这个家伙很懒，什么也没留下！

Tags | 热门标签

RankList | 热门文章