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

完整教程：大数据Spark（七十六）：Action行动算子reduce和take、takeSample使用案例

news 2026/3/26 18:39:05

文章目录

Action行动算子reduce和take、takeSample使用案例

一、reduce使用案例

二、take(n)使用案例

三、takeSample使用案例

Action行动算子reduce和take、takeSample使用案例

一、reduce使用案例

reduce可以对RDD中所有元素进行聚合操作，最终形成一个单一的结果。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("CustomPartitionerTest");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD rdd = sc.parallelize(Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10));
Integer reduce = rdd.reduce((a, b) -> a + b);
System.out.println(reduce);
sc.stop();

Scala代码：

val conf = new SparkConf().setMaster("local").setAppName("filter")
val sc = new SparkContext(conf)
val rdd: RDD[Int] = sc.parallelize(List(1, 2, 3, 4, 5, 6, 7, 8, 9, 10))
val result: Int = rdd.reduce(_ + _)
println(result)
sc.stop()

二、take(n)使用案例

返回一个包含数据集前n个元素的集合，结果会返回到Driver端。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("TakeTest");
JavaSparkContext sc = new JavaSparkContext(conf);
//take:返回RDD中的前n个元素
List takes = sc.parallelize(Arrays.asList("a", "b", "c", "d")).take(3);
System.out.println(takes.toString());
sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("TakeTest")
val sc = new SparkContext(conf)
//take: 取出RDD中前n个元素
val nums: Array[Int] = sc.parallelize(Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)).take(3)
println(nums.mkString(","))
sc.stop()

三、takeSample使用案例

takeSample用于从RDD中随机抽取指定数量的元素，返回一个集合到Driver端。方法签名如下：

def takeSample(withReplacement: Boolean,num: Int,seed: Long = Utils.random.nextLong): Array[T]

其中，withReplacement 指定抽样时是否放回，true表示有放回抽样，同一个元素可能被多次抽取，false表示无放回抽样，同一个元素最多被抽取一次;num 指定要随机抽取的元素数量；seed 是随机数生成器的种子，默认为随机生成，该参数如果固定，那么每次获取的数据固定。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("takeSampleTest");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaRDD rdd = sc.parallelize(Arrays.asList("a", "b", "c", "d", "e", "f","h","i","j","k"));
//takeSample: 从RDD中随机抽取num个元素，withReplacement表示是否放回，seed表示随机数种子
List list = rdd.takeSample(true, 4,100);
for (String s : list) {System.out.println(s);
}
sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("ForeachPartitionTest")
val sc = new SparkContext(conf)
val rdd: RDD[String] = sc.parallelize(List("a", "b", "c", "d", "e", "f", "h", "i", "j", "k"))
//takeSample: 从RDD中随机抽取num个元素，withReplacement表示是否放回，seed表示随机数种子
val strings: Array[String] = rdd.takeSample(true, 4, 100)
println(strings.mkString(","))
sc.stop()