@Test public void hashingBalancedPartitionerDoesBalance() { List<Double> reds = Arrays.asList(1.714D, 0.429D, .857D); List<Double> blues = Arrays.asList(0.9D, 0.6D, 1.5D); List<List<Double>> partitionWeights = Arrays.asList(reds, blues); HashingBalancedPartitioner hbp = new HashingBalancedPartitioner(partitionWeights); List<Tuple2<Integer, String>> l = new ArrayList<>(); for (int i = 0; i < 4; i++) { l.add(new Tuple2<Integer, String>(0, "red")); } for (int i = 0; i < 3; i++) { l.add(new Tuple2<Integer, String>(0, "blue")); } for (int i = 0; i < 1; i++) { l.add(new Tuple2<Integer, String>(1, "red")); } for (int i = 0; i < 2; i++) { l.add(new Tuple2<Integer, String>(1, "blue")); } for (int i = 0; i < 2; i++) { l.add(new Tuple2<Integer, String>(2, "red")); } for (int i = 0; i < 5; i++) { l.add(new Tuple2<Integer, String>(2, "blue")); } JavaPairRDD<Integer, String> rdd = JavaPairRDD.fromJavaRDD(sc.parallelize(l)).partitionBy(new HashPartitioner(3)); JavaPairRDD<Tuple2<Long, Integer>, String> indexedRDD = rdd.zipWithUniqueId().mapToPair( new PairFunction<Tuple2<Tuple2<Integer, String>, Long>, Tuple2<Long, Integer>, String>() { @Override public Tuple2<Tuple2<Long, Integer>, String> call( Tuple2<Tuple2<Integer, String>, Long> payLoadNuid) { Long uid = payLoadNuid._2(); String value = payLoadNuid._1()._2(); Integer elemClass = value.equals("red") ? 0 : 1; return new Tuple2<Tuple2<Long, Integer>, String>( new Tuple2<Long, Integer>(uid, elemClass), value); } }); List<Tuple2<Tuple2<Long, Integer>, String>> testList = indexedRDD.collect(); int[][] colorCountsByPartition = new int[3][2]; for (final Tuple2<Tuple2<Long, Integer>, String> val : testList) { Integer partition = hbp.getPartition(val._1()); if (val._2().equals("red")) colorCountsByPartition[partition][0] += 1; else colorCountsByPartition[partition][1] += 1; } for (int i = 0; i < 3; i++) { assertTrue(colorCountsByPartition[i][0] >= 1 && colorCountsByPartition[i][0] < 4); assertTrue(colorCountsByPartition[i][1] >= 2 && colorCountsByPartition[i][1] < 5); } }
@Override public int getPartition(Object key) { checkArgument(key instanceof Tuple2, "The key should be in the form: Tuple2(SparkUID, class) ..."); Tuple2<Long, Integer> uidNclass = (Tuple2<Long, Integer>) key; Long uid = uidNclass._1(); Integer partitionId = (int) (uid % numPartitions); Integer elementClass = uidNclass._2(); Double jumpProbability = Math.max(1D - 1D / partitionWeightsByClass.get(elementClass).get(partitionId), 0D); LinearCongruentialGenerator rand = new LinearCongruentialGenerator(uid); Double thisJumps = rand.nextDouble(); Integer thisPartition = partitionId; if (thisJumps < jumpProbability) { List<Double> jumpsTo = jumpTable.get(elementClass); Double destination = rand.nextDouble(); Integer probe = 0; while (jumpsTo.get(probe) < destination) { probe++; } thisPartition = probe; } return thisPartition; }
HashingBalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { checkArgument(key instanceof Tuple2, "The key should be in the form: Tuple2(SparkUID, class) ..."); Tuple2<Long, Integer> uidNclass = (Tuple2<Long, Integer>) key; Long uid = uidNclass._1(); Integer partitionId = (int) (uid % numPartitions); Integer elementClass = uidNclass._2(); Double jumpProbability = Math.max(1D - 1D / partitionWeightsByClass.get(elementClass).get(partitionId), 0D); LinearCongruentialGenerator rand = new LinearCongruentialGenerator(uid); Double thisJumps = rand.nextDouble(); Integer thisPartition = partitionId; if (thisJumps < jumpProbability) { List<Double> jumpsTo = jumpTable.get(elementClass); Double destination = rand.nextDouble(); Integer probe = 0; while (jumpsTo.get(probe) < destination) { probe++; } thisPartition = probe; } return thisPartition; } }
HashingBalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { checkArgument(key instanceof Tuple2, "The key should be in the form: Tuple2(SparkUID, class) ..."); Tuple2<Long, Integer> uidNclass = (Tuple2<Long, Integer>) key; Long uid = uidNclass._1(); Integer partitionId = (int) (uid % numPartitions); Integer elementClass = uidNclass._2(); Double jumpProbability = Math.max(1D - 1D / partitionWeightsByClass.get(elementClass).get(partitionId), 0D); LinearCongruentialGenerator rand = new LinearCongruentialGenerator(uid); Double thisJumps = rand.nextDouble(); Integer thisPartition = partitionId; if (thisJumps < jumpProbability) { List<Double> jumpsTo = jumpTable.get(elementClass); Double destination = rand.nextDouble(); Integer probe = 0; while (jumpsTo.get(probe) < destination) { probe++; } thisPartition = probe; } return thisPartition; } HashingBalancedPartitioner(List<List<Double>> partitionWeightsByClass); }
HashingBalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { checkArgument(key instanceof Tuple2, "The key should be in the form: Tuple2(SparkUID, class) ..."); Tuple2<Long, Integer> uidNclass = (Tuple2<Long, Integer>) key; Long uid = uidNclass._1(); Integer partitionId = (int) (uid % numPartitions); Integer elementClass = uidNclass._2(); Double jumpProbability = Math.max(1D - 1D / partitionWeightsByClass.get(elementClass).get(partitionId), 0D); LinearCongruentialGenerator rand = new LinearCongruentialGenerator(uid); Double thisJumps = rand.nextDouble(); Integer thisPartition = partitionId; if (thisJumps < jumpProbability) { List<Double> jumpsTo = jumpTable.get(elementClass); Double destination = rand.nextDouble(); Integer probe = 0; while (jumpsTo.get(probe) < destination) { probe++; } thisPartition = probe; } return thisPartition; } HashingBalancedPartitioner(List<List<Double>> partitionWeightsByClass); @Override int numPartitions(); @Override int getPartition(Object key); }
HashingBalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { checkArgument(key instanceof Tuple2, "The key should be in the form: Tuple2(SparkUID, class) ..."); Tuple2<Long, Integer> uidNclass = (Tuple2<Long, Integer>) key; Long uid = uidNclass._1(); Integer partitionId = (int) (uid % numPartitions); Integer elementClass = uidNclass._2(); Double jumpProbability = Math.max(1D - 1D / partitionWeightsByClass.get(elementClass).get(partitionId), 0D); LinearCongruentialGenerator rand = new LinearCongruentialGenerator(uid); Double thisJumps = rand.nextDouble(); Integer thisPartition = partitionId; if (thisJumps < jumpProbability) { List<Double> jumpsTo = jumpTable.get(elementClass); Double destination = rand.nextDouble(); Integer probe = 0; while (jumpsTo.get(probe) < destination) { probe++; } thisPartition = probe; } return thisPartition; } HashingBalancedPartitioner(List<List<Double>> partitionWeightsByClass); @Override int numPartitions(); @Override int getPartition(Object key);  }