// 传统的join操作会导致shuffle操作。// 因为两个RDD中，相同的key都需要通过网络拉取到一个节点上，由一个task进行join操作。val rdd3 = rdd1.join(rdd2)// Broadcast+map的join操作，不会导致shuffle操作。// 使用Broadcast将一个数据量较小的RDD作为广播变量。val rdd2Data = rdd2.collect()val rdd2DataBroadcast = sc.broadcast(rdd2Data)// 在rdd1.map算子中，可以从rdd2DataBroadcast中，获取rdd2的所有数据。// 然后进行遍历，如果发现rdd2中某条数据的key与rdd1的当前数据的key是相同的，//那么就判定可以进行join。// 此时就可以根据自己需要的方式，将rdd1当前数据与rdd2中可以连接的数据，//拼接在一起（String或Tuple）。val rdd3 = rdd1.map(rdd2DataBroadcast...)// 注意，以上操作，建议仅仅在rdd2的数据量比较少（比如几百M，或者一两G）的情况下使用。// 因为每个Executor的内存中，都会驻留一份rdd2的全量数据。

// 创建SparkConf对象。val conf = new SparkConf().setMaster(...).setAppName(...)// 设置序列化器为KryoSerializer。conf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")// 注册要序列化的自定义类型。conf.registerKryoClasses(Array(classOf[MyClass1], classOf[MyClass2]))

val conf = new SparkConf()val sc = new SparkContext(conf)val lines = sc.textFile("[hdfs://...](hdfs://...)")val words = lines.flatMap(_.split(" "))val pairs = words.map((_, 1))val wordCounts = pairs.reduceByKey(_ + _)wordCounts.collect().foreach(println(_))

// 第一步，给RDD中的每个key都打上一个随机前缀。JavaPairRDDLong> randomPrefixRdd = rdd.mapToPair(new PairFunctionLong,Long>, String, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2Long> call(Tuple2<Long, Long> tuple)throws Exception {  Random random = new Random();  int prefix = random.nextInt(10);  return new Tuple2Long>(prefix + "_" + tuple._1, tuple._2);}});// 第二步，对打上随机前缀的key进行局部聚合。JavaPairRDDLong> localAggrRdd = randomPrefixRdd.reduceByKey(new Function2<Long, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {  return v1 + v2;}});// 第三步，去除RDD中每个key的随机前缀。JavaPairRDD<Long, Long> removedRandomPrefixRdd = localAggrRdd.mapToPair(new PairFunctionLong>, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Long, Long> call(Tuple2Long> tuple)throws Exception {  long originalKey = Long.valueOf(tuple._1.split("_")[1]);  return new Tuple2<Long, Long>(originalKey, tuple._2);}});// 第四步，对去除了随机前缀的RDD进行全局聚合。JavaPairRDD<Long, Long> globalAggrRdd = removedRandomPrefixRdd.reduceByKey(new Function2<Long, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {  return v1 + v2;}});

// 首先从包含了少数几个导致数据倾斜key的rdd1中，采样10%的样本数据。JavaPairRDD<Long, String> sampledRDD = rdd1.sample(false, 0.1);// 对样本数据RDD统计出每个key的出现次数，并按出现次数降序排序。// 对降序排序后的数据，取出top 1或者top 100的数据，也就是key最多的前n个数据。// 具体取出多少个数据量最多的key，由大家自己决定，我们这里就取1个作为示范。JavaPairRDD<Long, Long> mappedSampledRDD = sampledRDD.mapToPair(new PairFunctionLong,String>, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Long, Long> call(Tuple2<Long, String> tuple)throws Exception {return new Tuple2<Long, Long>(tuple._1, 1L);}    });JavaPairRDD<Long, Long> countedSampledRDD = mappedSampledRDD.reduceByKey(new Function2<Long, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Long call(Long v1, Long v2) throws Exception {return v1 + v2;}});JavaPairRDD<Long, Long> reversedSampledRDD = countedSampledRDD.mapToPair(new PairFunctionLong,Long>, Long, Long>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Long, Long> call(Tuple2<Long, Long> tuple)throws Exception {return new Tuple2<Long, Long>(tuple._2, tuple._1);}});final Long skewedUserid = reversedSampledRDD.sortByKey(false).take(1).get(0)._2;// 从rdd1中分拆出导致数据倾斜的key，形成独立的RDD。JavaPairRDD<Long, String> skewedRDD = rdd1.filter(new FunctionLong,String>, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2<Long, String> tuple) throws Exception {return tuple._1.equals(skewedUserid);}});// 从rdd1中分拆出不导致数据倾斜的普通key，形成独立的RDD。JavaPairRDD<Long, String> commonRDD = rdd1.filter(new FunctionLong,String>, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2<Long, String> tuple) throws Exception {return !tuple._1.equals(skewedUserid);}});// rdd2，就是那个所有key的分布相对较为均匀的rdd。// 这里将rdd2中，前面获取到的key对应的数据，过滤出来，分拆成单独的rdd，并对rdd中的数据使用flatMap算子都扩容100倍。// 对扩容的每条数据，都打上0～100的前缀。JavaPairRDD skewedRdd2 = rdd2.filter(new FunctionLong,Row>, Boolean>() {private static final long serialVersionUID = 1L;@Overridepublic Boolean call(Tuple2<Long, Row> tuple) throws Exception {return tuple._1.equals(skewedUserid);}}).flatMapToPair(new PairFlatMapFunctionLong,Row>, String, Row>() {private static final long serialVersionUID = 1L;@Overridepublic Iterable> call(Tuple2<Long, Row> tuple) throws Exception {Random random = new Random();List> list = new ArrayList>();for(int i = 0; i < 100; i++) {list.add(new Tuple2(i + "_" + tuple._1, tuple._2));}return list;}});// 将rdd1中分拆出来的导致倾斜的key的独立rdd，每条数据都打上100以内的随机前缀。// 然后将这个rdd1中分拆出来的独立rdd，与上面rdd2中分拆出来的独立rdd，进行join。JavaPairRDD<Long, Tuple2> joinedRDD1 = skewedRDD.mapToPair(new PairFunctionLong,String>, String, String>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2 call(Tuple2<Long, String> tuple)throws Exception {Random random = new Random();int prefix = random.nextInt(100);return new Tuple2(prefix + "_" + tuple._1, tuple._2);}}).join(skewedUserid2infoRDD).mapToPair(new PairFunction>, Long, Tuple2>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2<Long, Tuple2> call(Tuple2> tuple)throws Exception {long key = Long.valueOf(tuple._1.split("_")[1]);return new Tuple2<Long, Tuple2>(key, tuple._2);}});// 将rdd1中分拆出来的包含普通key的独立rdd，直接与rdd2进行join。JavaPairRDD<Long, Tuple2> joinedRDD2 = commonRDD.join(rdd2);// 将倾斜key join后的结果与普通key join后的结果，uinon起来。// 就是最终的join结果。JavaPairRDD<Long, Tuple2> joinedRDD = joinedRDD1.union(joinedRDD2);

// 首先将其中一个key分布相对较为均匀的RDD膨胀100倍。JavaPairRDD<String, Row> expandedRDD = rdd1.flatMapToPair(new PairFlatMapFunction, String, Row>() {private static final long serialVersionUID = 1L;@Overridepublic IterableString, Row>> call(Tuple2 tuple)throws Exception {ListString, Row>> list = new ArrayListString, Row>>();for(int i = 0; i < 100; i++) {list.add(new Tuple2<String, Row>(0 + "_" + tuple._1, tuple._2));}return list;}});// 其次，将另一个有数据倾斜key的RDD，每条数据都打上100以内的随机前缀。JavaPairRDD<String, String> mappedRDD = rdd2.mapToPair(new PairFunctionString>, String, String>() {private static final long serialVersionUID = 1L;@Overridepublic Tuple2<String, String> call(Tuple2String> tuple)throws Exception {Random random = new Random();int prefix = random.nextInt(100);return new Tuple2<String, String>(prefix + "_" + tuple._1, tuple._2);}});// 将两个处理后的RDD进行join即可。JavaPairRDD<String, Tuple2<String, Row>> joinedRDD = mappedRDD.join(expandedRDD);