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

广州网站建设企业化妆品网络营销策划方案

news 2025/8/4 20:53:59
广州网站建设企业,化妆品网络营销策划方案,宁夏做网站公司,广州网站制作重分区函数 如何对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.yidumall.com/news/69068.html

相关文章：

  • 邯郸哪做网站如何免费推广一个网站
  • 网站设计与网站建设a卷谷歌网址
  • 网站开发的开发语言如何做网站推广及优化
  • 深圳网站建设.-方维网络网易企业邮箱
  • 微商引流推广平台营销网站优化推广
  • 深圳宝安做网站公司搜索引擎优化的概念是什么
  • 福州网站建设机构衡阳seo外包
  • 公众号网站制作网址搜索ip地址
  • 工程承包去哪个网站活动营销方案
  • 贵州政务网站建设规范东营网站建设费用
  • 网站建设技术网站建设最新军事动态
  • 福州网站怎么做seo中国营销策划第一人
  • 汝城网站建设公司关键词查询工具免费
  • 北京房地产网站建设软文代发代理
  • 定制网站建设多少钱海淀搜索引擎优化seo
  • 东莞网站设计如何优化排名seo
  • 网站美工用什么软件优化网站排名解析推广
  • 河间哪里有做网站的百度指数pc版
  • 做外贸必须有网站吗优化营商环境工作总结
  • 织梦cms发布侵权网站清单武汉seo网站优化技巧
  • 企业网站源码带后台管理站长之家查询网站
  • 做网站在图片里加文字抚顺网络推广
  • 哔哩哔哩网站提升关键词排名seo软件
  • vue做前台网站百度官方人工客服电话
  • 教育网站制作运营怎么做网络平台
  • 网站建设费用都包括什么科目永久免费的建站系统有哪些
  • 免费网站制作网站源码网页开发用什么软件
  • 成都网站建设全平台黑帽seo技巧
  • 网站建设愿景游戏推广
  • 单页面组合网站网络营销的原理