用spark的算子aggregateByKey举例

使用Spark的算子aggregateByKey举例，可以参考以下代码：

import org.apache.spark.{SparkConf, SparkContext}

object Demo {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local").setAppName("aa")
    val sc = new SparkContext(conf)
    
    // 创建一个包含键值对的列表
    val list = List((1,2),(2,7),(1,3),(2,8),(3,9),(3,10),(1,4),(1,5),(2,6),(2,11),(3,12),(3,13))
    
    // 将列表转化为RDD
    val listRDD = sc.parallelize(list,2)
    
    // 使用aggregateByKey算子计算每个键对应的最大值
    val result = listRDD.aggregateByKey(0)(
      (a, b) => math.max(a, b),
      (x, y) => math.max(x, y)
    )
    
    // 打印结果
    result.collect().foreach(println)
  }
}

以上代码中，我们创建了一个包含键值对的列表，然后将其转化为RDD。接下来，使用aggregateByKey算子计算每个键对应的最大值。在seqOp函数中，我们比较当前值与累加器并返回较大的值。在combOp函数中，我们再次比较两个累加器的值并返回较大的值。最后，通过collect函数将结果打印出来。<span class="em">1</span><span class="em">2</span><span class="em">3</span>
#### 引用[.reference_title]
- *1* *2* *3* [详解Spark核心算子 : aggregateByKey和combineByKey](https://blog.csdn.net/f_n_c_k/article/details/88718262)[target="_blank" data-report-click={"spm":"1018.2226.3001.9630","extra":{"utm_source":"vip_chatgpt_common_search_pc_result","utm_medium":"distribute.pc_search_result.none-task-cask-2~all~insert_cask~default-1-null.142^v93^chatsearchT3_2"}}] [.reference_item style="max-width: 100%"]
[ .reference_list ]