当前位置： 首页 > news >正文

广州网站建设企业上海培训机构排名榜

news 2025/7/4 21:11:52
广州网站建设企业,上海培训机构排名榜,做时彩网站违法吗,wordpress机械模板下载地址重分区函数 如何对RDD中分区数目进行调整（增加分区或减少分区），在RDD函数中主要有如下三个函数。 1）、增加分区函数 函数名称：repartition，此函数使用的谨慎，会产生Shuffle。 2）、…

重分区函数

如何对RDD中分区数目进行调整（增加分区或减少分区），在RDD函数中主要有如下三个函数。
1）、增加分区函数
函数名称：repartition，此函数使用的谨慎，会产生Shuffle。
在这里插入图片描述
2）、减少分区函数
函数名称：coalesce，此函数不会产生Shuffle，当且仅当降低RDD分区数目。
比如RDD的分区数目为10个分区，此时调用rdd.coalesce(12)，不会对RDD进行任何操作。
在这里插入图片描述
3）、调整分区函数
在PairRDDFunctions（此类专门针对RDD中数据类型为KeyValue对提供函数）工具类中
partitionBy函数：
在这里插入图片描述
范例演示代码，适当使用函数调整RDD分区数目：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* RDD中分区函数，调整RDD分区数目，可以增加分区和减少分区
*/
object SparkPartitionTest {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 读取本地文件系统文本文件数据
val datasRDD: RDD[String] = sc.textFile("datas/wordcount/wordcount.data", minPartitions = 2)
// TODO: 增加RDD分区数
val etlRDD: RDD[String] = datasRDD.repartition(3)
println(s"EtlRDD 分区数目 = ${etlRDD.getNumPartitions}")
// 词频统计
val resultRDD: RDD[(String, Int)] = etlRDD
// 数据分析，考虑过滤脏数据
.filter(line => null != line && line.trim.length > 0)
// 分割单词，注意去除左右空格
.flatMap(line => line.trim.split("\\s+"))
// 转换为二元组，表示单词出现一次
.mapPartitions{iter =>
iter.map(word => (word, 1))
}
// 分组聚合，按照Key单词
.reduceByKey((tmp, item) => tmp + item)
// 输出结果RDD
resultRDD
// TODO: 对结果RDD降低分区数目
.coalesce(1)
.foreachPartition(iter => iter.foreach(println))
// 应用程序运行结束，关闭资源
sc.stop()
}
}

在实际开发中，什么时候适当调整RDD的分区数目呢？让程序性能更好好呢？？？？

第一点：增加分区数目

  • 当处理的数据很多的时候，可以考虑增加RDD的分区数目
    在这里插入图片描述

第二点：减少分区数目

  • 其一：当对RDD数据进行过滤操作（filter函数）后，考虑是否降低RDD分区数目
    在这里插入图片描述
  • 其二：当对结果RDD存储到外部系统
    在这里插入图片描述

聚合函数

在数据分析领域中，对数据聚合操作是最为关键的，在Spark框架中各个模块使用时，主要就是其中聚合函数的使用。

集合中聚合函数
回顾列表List中reduce聚合函数核心概念：聚合的时候，往往需要聚合中间临时变量。查看列表List中聚合函数reduce和fold源码如下：
在这里插入图片描述
通过代码，看看列表List中聚合函数使用：
在这里插入图片描述
运行截图如下所示：
在这里插入图片描述
fold聚合函数，比reduce聚合函数，多提供一个可以初始化聚合中间临时变量的值参数：
在这里插入图片描述
聚合操作时，往往聚合过程中需要中间临时变量（到底时几个变量，具体业务而定），如下案例：
在这里插入图片描述

RDD 中聚合函数
在RDD中提供类似列表List中聚合函数reduce和fold，查看如下：
在这里插入图片描述
案例演示：求列表List中元素之和，RDD中分区数目为2，核心业务代码如下：
在这里插入图片描述
运行原理分析：
在这里插入图片描述
使用RDD中fold聚合函数：
在这里插入图片描述
查看RDD中高级聚合函数aggregate，函数声明如下：
在这里插入图片描述
业务需求：使用aggregate函数实现RDD中最大的两个数据，分析如下：
在这里插入图片描述
核心业务代码如下：
在这里插入图片描述
运行结果原理剖析示意图：
在这里插入图片描述

上述完整范例演示代码：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext, TaskContext}
import scala.collection.mutable
import scala.collection.mutable.ListBuffer
/**
* RDD中聚合函数：reduce、aggregate函数
*/
object SparkAggTest {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 模拟数据，1 到 10 的列表，通过并行方式创建RDD
val datas = 1 to 10
val datasRDD: RDD[Int] = sc.parallelize(datas, numSlices = 2)
// 查看每个分区中的数据
datasRDD.foreachPartition{iter =>
println(s"p-${TaskContext.getPartitionId()}: ${iter.mkString(", ")}")
}
println("=========================================")
// 使用reduce函数聚合
val result: Int = datasRDD.reduce((tmp, item) => {
println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
tmp + item
})
println(result)
println("=========================================")
// 使用fold函数聚合
val result2: Int = datasRDD.fold(0)((tmp, item) => {
println(s"p-${TaskContext.getPartitionId()}: tmp = $tmp, item = $item")
tmp + item
})
println(result2)
println("=========================================")
// 使用aggregate函数获取最大的两个值
val top2: mutable.Seq[Int] = datasRDD.aggregate(new ListBuffer[Int]())(
// 分区内聚合函数，每个分区内数据如何聚合 seqOp: (U, T) => U,
(u, t) => {
println(s"p-${TaskContext.getPartitionId()}: u = $u, t = $t")
// 将元素加入到列表中
u += t //
// 降序排序
val top = u.sorted.takeRight(2)
// 返回
top
},
// 分区间聚合函数，每个分区聚合的结果如何聚合 combOp: (U, U) => U
(u1, u2) => {
println(s"p-${TaskContext.getPartitionId()}: u1 = $u1, u2 = $u2")
u1 ++= u2 // 将列表的数据合并，到u1中
//
u1.sorted.takeRight(2)
}
)
println(top2)
// 应用程序运行结束，关闭资源
sc.stop()
}
}

PairRDDFunctions 聚合函数
在Spark中有一个object对象PairRDDFunctions，主要针对RDD的数据类型是Key/Value对的数据提供函数，方便数据分析处理。比如使用过的函数：reduceByKey、groupByKey等。*ByKey函数：将相同Key的Value进行聚合操作的，省去先分组再聚合。

第一类：分组函数groupByKey
在这里插入图片描述
第二类：分组聚合函数reduceByKey和foldByKey
在这里插入图片描述
但是reduceByKey和foldByKey聚合以后的结果数据类型与RDD中Value的数据类型是一样的。

第三类：分组聚合函数aggregateByKey
在这里插入图片描述
在企业中如果对数据聚合使用，不能使用reduceByKey完成时，考虑使用aggregateByKey函数，基本上都能完成任意聚合功能。

演示范例代码如下：

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
/**
* RDD中聚合函数，针对RDD中数据类型Key/Value对：
* groupByKey
* reduceByKey/foldByKey
* aggregateByKey
* combineByKey
*/
object SparkAggByKeyTest {
def main(args: Array[String]): Unit = {
// 创建应用程序入口SparkContext实例对象
val sc: SparkContext = {
// 1.a 创建SparkConf对象，设置应用的配置信息
val sparkConf: SparkConf = new SparkConf()
.setAppName(this.getClass.getSimpleName.stripSuffix("$"))
.setMaster("local[2]")
// 1.b 传递SparkConf对象，构建Context实例
new SparkContext(sparkConf)
}
sc.setLogLevel("WARN")
// 1、并行化集合创建RDD数据集
val linesSeq: Seq[String] = Seq(
"hadoop scala hive spark scala sql sql", //
"hadoop scala spark hdfs hive spark", //
"spark hdfs spark hdfs scala hive spark" //
)
val inputRDD: RDD[String] = sc.parallelize(linesSeq, numSlices = 2)
// 2、分割单词，转换为二元组
val wordsRDD: RDD[(String, Int)] = inputRDD
.flatMap(line => line.split("\\s+"))
.map(word => word -> 1)
// TODO: 先使用groupByKey函数分组，再使用map函数聚合
val wordsGroupRDD: RDD[(String, Iterable[Int])] = wordsRDD.groupByKey()
val resultRDD: RDD[(String, Int)] = wordsGroupRDD.map{ case (word, values) =>
val count: Int = values.sum
word -> count
}
println(resultRDD.collectAsMap())
// TODO: 直接使用reduceByKey或foldByKey分组聚合
val resultRDD2: RDD[(String, Int)] = wordsRDD.reduceByKey((tmp, item) => tmp + item)
println(resultRDD2.collectAsMap())
val resultRDD3 = wordsRDD.foldByKey(0)((tmp, item) => tmp + item)
println(resultRDD3.collectAsMap())
// TODO: 使用aggregateByKey聚合
/*
def aggregateByKey[U: ClassTag]
(zeroValue: U) // 聚合中间临时变量初始值，类似fold函数zeroValue
(
seqOp: (U, V) => U, // 各个分区内数据聚合操作函数
combOp: (U, U) => U // 分区间聚合结果的聚合操作函数
): RDD[(K, U)]
*/
val resultRDD4 = wordsRDD.aggregateByKey(0)(
(tmp: Int, item: Int) => {
tmp + item
},
(tmp: Int, result: Int) => {
tmp + result
}
)
println(resultRDD4.collectAsMap())
// 应用程序运行结束，关闭资源
Thread.sleep(1000000)
sc.stop()
}
}

面试题
RDD中groupByKey和reduceByKey区别？？？

  • reduceByKey函数：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的reduce函数，将相同key的值聚合到一起，reduce任务的个数可以通过第二个可选的参数来设置。
    在这里插入图片描述
  • groupByKey函数：在一个(K,V)的RDD上调用，返回一个(K,V)的RDD，使用指定的函数，将相同key的值聚合到一起，与reduceByKey的区别是只生成一个sequence。
    在这里插入图片描述
http://www.dt0577.cn/news/1905.html

相关文章：

  • 网站外部链接做多少合适呢搜索引擎调词工具
  • 大型网站制作女教师遭网课入侵直播录屏曝光i
  • 网站开发毕业生报告seowhy论坛
  • 做网站公司在哪seo推广优化服务
  • 张家口企业网站建设怎样策划一个营销型网站
  • wordpress建站必须选择主题百度搜一下
  • 下载搭建网站软件下载中国国家人事人才培训网证书查询
  • 淄博软件开发公司有哪些石家庄seo优化
  • 北京企业网站备案厦门seo屈兴东
  • 旅游网站策划书网站设计制作在哪能看
  • 网络营销推广方法工具seo排名关键词点击
  • 国内有哪些做卡通素材的网站网站优化排名方法
  • 马上飞做的一些网站今天
  • 做建站较好的网站免费域名注册二级域名
  • 用什么网站做问卷企业推广app
  • wap网站建设管理制度站长之家域名查询排行
  • 广东装饰公司网站建设seo哪家公司好
  • 百度怎样做网站排名往前整站seo技术搜索引擎优化
  • 新手如何做网站电商网站设计论文
  • 年终总结汇报ppt模板sem推广优化
  • 网站怎么做自然优化游戏推广合作平台
  • php做网站的公司有哪些新闻热点事件2021(最新)
  • 阿里巴巴1688批发网济南seo网站优化公司
  • 网站建设在开封找谁做免费b2b推广网站大全
  • 网络建设公司有哪些合肥seo推广外包
  • 怎么做flash网站设计友情链接的形式有哪些
  • 视频网站发展好应该怎么做中国四大软件外包公司
  • 重庆合川企业网站建设联系电话软文免费发布平台
  • 怎么用源码做网站视频宁波seo外包公司
  • 做电商网站需要多少钱企业网站seo托管怎么做