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

深入解析：大数据Spark（六十六）：Transformation转换算子sample、sortBy和sortByKey

news 2025/10/21 19:25:31

深入解析：大数据Spark（六十六）：Transformation转换算子sample、sortBy和sortByKey

文章目录

Transformation转换算子sample、sortBy和sortByKey

一、sample

二、sortBy

三、sortByKey

Transformation转换算子sample、sortBy和sortByKey

一、sample

随机抽样算子，根据传进去的小数按比例进行有放回或者无放回的抽样，常用于数据预览、测试或处理大规模数据时的抽样分析。sample算子函数签名如下：

def sample(
    withReplacement: Boolean,
    fraction: Double,
    seed: Long = Utils.random.nextLong
): RDD[T]

withReplacement：布尔值，表示抽样时是否采用有放回的方式。true 表示有放回抽样，即同一个元素可能被多次抽取；false 表示无放回抽样，每个元素最多被抽取一次。
fraction：表示抽样比例。对于无放回抽样，fraction 是期望抽取的样本占原始数据集的比例，取值范围为 [0, 1]；对于有放回抽样，fraction 表示每个元素被抽取的期望次数，取值应大于等于 0。
seed：可选参数，表示随机数生成器的种子，指定种子可以确保每次抽样结果一致，便于调试和测试。

需求：对数据进行有放回抽样。

Java代码

SparkConf conf = new SparkConf().setMaster("local").setAppName("sample");
JavaSparkContext sc = new JavaSparkContext(conf);
//sample(withReplacement, fraction, seed), withReplacement表示是否放回，fraction表示采样比例，seed表示随机种子
sc.parallelize(Arrays.asList(1,2,3,4,5,6,7,8,9,10)).sample(false,0.5,10).foreach(s -> System.out.println(s));
sc.stop();

Scala代码

val conf = new SparkConf().setMaster("local").setAppName("SampleTest")
val sc = new SparkContext(conf)
//sample(withReplacement, fraction, seed),
//withReplacement表示是否放回，fraction表示抽样的比例，seed表示随机种子
sc.parallelize(1 to 10).sample(false, 0.5,10).foreach(println)
sc.stop()

二、sortBy

sortBy 对任意类型的RDD 中的元素按照指定的键进行排序，需要一个函数来提取排序键，并可以指定升序或降序，以及分区数。函数签名如下：

def sortBy[K](f: (T) => K,ascending: Boolean = true,numPartitions: Int = this.partitions.length
)(implicit ord: Ordering[K], ctag: ClassTag[K]): RDD[T]

f：从 RDD 的元素中提取排序键的函数。
ascending：布尔值，表示是否按升序排序，默认为 true。
numPartitions：排序后 RDD 的分区数，默认为与原始 RDD 相同。

特别注意：Java API中K,V格式RDD没有sortBy操作。

案例：按照字符串长度降序排序。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("SortByTest");
JavaSparkContext sc = new JavaSparkContext(conf);
sc.parallelize(Arrays.asList("zhangsan", "lisi", "wangwu", "maliu"))//sortBy:按照字符串长度进行排序.sortBy(new Function() {@Overridepublic Integer call(String s) throws Exception {return s.length();}},true,1).foreach(s -> System.out.println(s ));
sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("SortByTest")
val sc = new SparkContext(conf)
//sortBy(f, ascending, numPartitions)
//f表示排序的依据，ascending表示是否升序，numPartitions表示分区数
sc.parallelize(List("zhangsan","lisi","wangwu","maliu")).sortBy(_.length).foreach(println)
sc.stop()

三、sortByKey

sortByKey 专用于对键值对 RDD 的键进行排序,可以指定升序或降序，以及分区数。

案例：按照K,V数据中的key大小降序排序。

Java代码：

SparkConf conf = new SparkConf().setMaster("local").setAppName("SortByKeyTest");
JavaSparkContext sc = new JavaSparkContext(conf);
JavaPairRDD rdd = sc.parallelizePairs(Arrays.asList(new Tuple2(10, "zhangsan"),new Tuple2(20, "lisi"),new Tuple2(30, "wangwu"),new Tuple2(40, "maliu")
));
//sortBy:按照字符串长度进行排序
//按照key进行排序，false降序排序
rdd.sortByKey(false).foreach(new VoidFunction>() {@Overridepublic void call(Tuple2 integerStringTuple2) throws Exception {System.out.println(integerStringTuple2);}
});
sc.stop();

Scala代码：

val conf: SparkConf = new SparkConf().setMaster("local").setAppName("SortByKeyTest")
val sc = new SparkContext(conf)
sc.parallelize(List(("a",1),("b",2),("c",3),("d",4)))//sortByKey(ascending, numPartitions)//ascending表示是否升序，numPartitions表示分区数.sortByKey(false).foreach(println)
sc.stop()