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

RDD的自定义分区器-案例

news 来源：原创 2025/6/22 22:42:19

对电商订单数据进行处理，订单数据包含用户 ID 和订单金额，不同地区的用户有不同的 ID 范围。我们会按照地区对订单数据进行分区，这样做能让相同地区的订单数据处于同一分区，便于后续按地区进行统计金额分析。

订单数据如下：

(500, 100.0),
(1200, 200.0),
(2500, 300.0),
(800, 150.0),
(1800, 250.0),
(2200, 350.0)

要求是：

0-1000号分成一个区；

1001-2000号分成一个区；

2001-2500号分成一个区；

步骤：

1. 数据加载

需要把给定的订单数据加载到 Spark 的 RDD（弹性分布式数据集）或者 DataFrame 中。在这个需求里，订单数据以键值对的形式存在，其中键是用户 ID，值是订单金额。可以使用 parallelize 方法把数据转换成 RDD。

2. 自定义分区器

由于默认的分区器无法满足按照用户 ID 范围分区的需求，所以要自定义一个分区器。这个分区器要依据用户 ID 的范围把订单数据分到不同的分区。具体来说，将用户 ID 在 0 - 1000 的订单数据分到一个分区，1001 - 2000 的分到另一个分区，2001 及以上的分到第三个分区。

3. 数据分区

使用自定义分区器对 RDD 进行分区操作，确保相同地区（用户 ID 范围相同）的订单数据处于同一分区。

4. 数据汇总

对每个分区内的订单数据进行汇总，计算每个地区的订单总金额。可以使用 reduceByKey 或者 aggregateByKey 等方法来实现汇总操作。

5. 结果输出

将汇总后的结果输出，展示每个地区的订单总金额。

功能实现：

创建新的maven项目。
创建input文件夹，在input下新建记事本文件，其中内容就是前面的实例数据。
在src下创建新的scala文件，开始写功能代码。

// 1. 实现自定义分区器
// 2. 读文件，生成RDD// 3. RDD使用自定义分区器分区// 4. 对分区的数据进行汇总计算// 5. 保存计算之后的结果

import org.apache.spark.{Partitioner, SparkConf, SparkContext}// 创建一个类继承Partitioner
class OrderPartitioner extends Partitioner {override def numPartitions: Int = 2 // 两个分区，编号就是： 0， 1// key - valueoverride def getPartition(key: Any): Int = {// 如果key在2001和2003之间，就返回 0// 否则，返回 1val keyInt = key.asInstanceOf[Int]if (keyInt > 2000 && keyInt < 2003) {0} else {1}}
}// case class
case class Order(id: Int, price: Double, category: String)object PartitionOrder {def main(args: Array[String]): Unit = {// 创建SparkContextval conf = new SparkConf().setAppName("Partition").setMaster("local[*]")val sc = new SparkContext(conf)// 初始数据val rdd = sc.textFile("data/order.csv")val rdd1 = rdd.map( line => {val fields = line.split(",")(fields(0).toInt, Order(fields(0).toInt, fields(1).toDouble, fields(2)))})// 使用自定义分区器val rdd2 = rdd1.partitionBy(new OrderPartitioner)rdd2.map( x => x._2).saveAsTextFile("output18")val regionTotalAmount = rdd2.mapPartitions((iter) => {var count = 0var totalAmount = 0.0// 同时计算件数和总金额while (iter.hasNext) {val item = iter.next()count += 1val price = item._2.priceprintln(price)totalAmount += price}Iterator(s"${count}件,$totalAmount")})// 在分区完成之后的基础上，只保留key
//    val rdd3 = rdd2.map( x => x._2)regionTotalAmount.saveAsTextFile("output19")}
}

相关文章：