。

想了解的朋友可以查看这篇文章。

同时#xff0c;希望我的文章能帮助到你#xff0c;如果觉得我的文章写的不错#xff0c;请留下你宝贵的点赞#xff0c;谢…在之前的文章中我们学习了如何在spark中使用键值对中的keys和values,reduceByKey,groupByKey三种方法。

想了解的朋友可以查看这篇文章。

同时希望我的文章能帮助到你如果觉得我的文章写的不错请留下你宝贵的点赞谢谢。

Spark-Scala语言实战12-CSDN博客文章浏览阅读722次点赞19次收藏15次。

今天开始的文章我会带给大家如何在spark的中使用我们的键值对方法今天学习键值对方法中的keys和values,reduceByKey,groupByKey三种方法。

希望我的文章能帮助到大家也欢迎大家来我的文章下交流讨论共同进步。

https://blog.csdn.net/qq_49513817/article/details/137385224今天的文章开始我会继续带着大家如何在spark的中使用我们的键值对里的方法。

今天学习键值对方法中的fullOuterJoinzipcombineByKey三种方法。

一、知识回顾

上一篇文章中我们学习了键值对的三种方法分别是keys和values,reduceByKey,groupByKey。

reduceByKey可以进行统计将有相同键的值进行相加统一输出。

二、键值对方法

fullOuterJoin()方法用于对两个RDD进行全外连接保留两个RDD中所有键的连接结果。

import

SparkConf().setMaster(local).setAppName(p2)val

new

收集结果并打印ppp.collect().foreach(println)}

fullOuterJoin方法全外连接那么两个键值对都会连接。

可以看到两个键值对里的键与值都连接上了互相没有的值即显示None值。

2.zip

zip()方法用于将两个RDD组合成键值对RDD要求两个RDD的分区数量以及元素数量相同否则会抛出异常。

将两个RDD组合成Key/Value形式的RDD这里要求两个RDD的partition数量以及元素数量都相同否则会抛出异常。

import

SparkConf().setMaster(local).setAppName(p2)val

new

收集结果并打印pp1.collect().foreach(println)pp2.collect().foreach(println)}

代码创建了两个不同的RDD键值对分别使用p1zip方法p2与p2zip方法p1那么它们输出的结果会是一样的吗

3.combineByKey

combineByKey()方法是Spark中一个比较核心的高级方法键值对的其他一些高级方法底层均是使用combineByKey()方法实现的如groupByKey()方法、reduceByKey()方法等。

combineByKey()方法用于将键相同的数据聚合并且允许返回类型与输入数据的类型不同的返回值。

combineByKey()方法的使用方式如下。

combineByKey(createCombiner,mergeValue,mergeCombiners,numPartitionsNone)

import

SparkConf().setMaster(local).setAppName(p2)val

new

这个函数定义了如何将新值与当前的累加器值合并。

在我的代码中我将新值与累加器相加。

mergeCombiners:

这个函数定义了当两个累加器对应于同一个键但可能来自不同的分区需要合并时应该执行的操作。

在我的代码中也是将两个累加器值相加

这确保了无论数据如何在分区之间分布最终每个键都会得到正确的累加结果。

看看输出效果

方法参数描述例子fullOuterJoino***rRDD另一个要与之进行全外连接的RDDrdd1.fullOuterJoin(rdd2)fullOuterJoinnumPartitions结果RDD的分区数可选rdd1.fullOuterJoin(rdd2,

numPartitions10)zipo***rRDD要与之进行zip操作的另一个RDDrdd1.zip(rdd2)combineByKeycreateCombiner处理第一个出现的每个键的值的函数lambda

(v,

1)combineByKeymergeValue合并具有相同键的值的函数lambda

acc,

1)combineByKeymergeCombiners合并两个累积器的函数lambda

acc1,

acc2[1])combineByKeynumPartitions结果RDD的分区数可选rdd.combineByKey(createCombiner,

mergeValue,