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

【大数据开发】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

推荐阅读

schema
Spring – Bean Life Cycle

Spring – Bean Life Cycle ... [详细]

蜡笔小新 2024-11-13 13:24:40
schema
Java设计模式详解：解释器模式的应用与实现

本文详细介绍了Java设计模式中的解释器模式，包括其定义、应用场景、优缺点以及具体的实现示例。通过音乐解释器的例子，帮助读者更好地理解和应用这一模式。 ... [详细]

蜡笔小新 2024-11-14 21:00:34
join
使用Python爬取妙笔阁小说信息并保存为TXT和CSV格式

本文介绍了如何使用Python爬取妙笔阁小说网仙侠系列中所有小说的信息，并将其保存为TXT和CSV格式。主要内容包括如何构造请求头以避免被网站封禁，以及如何利用XPath解析HTML并提取所需信息。 ... [详细]

蜡笔小新 2024-11-14 19:54:58
include
普通树(每个节点可以有任意数量的子节点)级序遍历

普通树(每个节点可以有任意数量的子节点)级序遍历 ... [详细]

蜡笔小新 2024-11-14 18:53:26
post
Go Echo 框架入门指南【1】

本文介绍了 Go 语言中的高性能、可扩展、轻量级 Web 框架 Echo。Echo 框架简单易用，仅需几行代码即可启动一个高性能 HTTP 服务。 ... [详细]

蜡笔小新 2024-11-14 18:30:58
default
将.o文件链接到.elf文件时

我有一个从C项目编译的.o文件，该文件引用了名为init_static_pool ... [详细]

蜡笔小新 2024-11-14 10:07:21
default
JVM钩子函数的应用场景详解

本文详细介绍了JVM钩子函数的多种应用场景，包括正常关闭、异常关闭和强制关闭。通过具体示例和代码演示，帮助读者更好地理解和应用这一机制。适合对Java编程和JVM有一定基础的开发者阅读。 ... [详细]

蜡笔小新 2024-11-13 18:34:48
default
HDFS API

Hadoop的文件操作位于包org.apache.hadoop.fs里面，能够进行新建、删除、修改等操作。比较重要的几个类：(1)Configurati ... [详细]

蜡笔小新 2024-11-13 17:31:50
default
Java反射机制详解及应用场景

本文详细介绍了Java反射机制的基本概念、获取Class对象的方法、反射的主要功能及其在实际开发中的应用。通过具体示例，帮助读者更好地理解和使用Java反射。 ... [详细]

蜡笔小新 2024-11-13 16:08:08
default
Java DAO模式详解与代码示例

DAO（Data Access Object）模式是一种用于抽象和封装所有对数据库或其他持久化机制访问的方法，它通过提供一个统一的接口来隐藏底层数据访问的复杂性。 ... [详细]

蜡笔小新 2024-11-13 12:25:33
grid
Leetcode学习成长记：天池leetcode基础训练营Task01数组

前言这是本人第一次参加由Datawhale举办的组队学习活动，这个活动每月一次，之前也一直关注，但未亲身参与过，这次看到活动 ... [详细]

蜡笔小新 2024-11-14 18:01:31
settings
Native与HTML5交互基础教程

本文将介绍如何在混合开发（Hybrid）应用中实现Native与HTML5的交互，包括基本概念、学习目标以及具体的实现步骤。 ... [详细]

蜡笔小新 2024-11-14 12:33:11
schema
Android Studio SQLite 数据库增删改查简单（代码参考）

一个建表一个执行crud操作建表代码importandroid.content.Context;importandroid.database.sqlite.SQLiteDat ... [详细]

蜡笔小新 2024-11-14 11:01:49
join
Python基础：使用NLTK和Python构建机器学习应用

本文节选自《NLTK基础教程——用NLTK和Python库构建机器学习应用》一书的第1章第1.2节，作者Nitin Hardeniya。本文将带领读者快速了解Python的基础知识，为后续的机器学习应用打下坚实的基础。 ... [详细]

蜡笔小新 2024-11-13 21:23:34
header
Cookie学习小结

Cookie学习小结 ... [详细]

蜡笔小新 2024-11-14 16:26:25

小木马

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

Tags | 热门标签

RankList | 热门文章