@Test public void balancedPartitionerFirstElementsWithRemainder() { BalancedPartitioner bp = new BalancedPartitioner(10, 10, 1); for (int i = 0; i < 10; i++) { int p = bp.getPartition(i); assertEquals("Found wrong partition output " + p + ", not 0", 0, p); } }
@Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } }
BalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } } }
BalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } } BalancedPartitioner(int numPartitions, int elementsPerPartition, int remainder); }
BalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } } BalancedPartitioner(int numPartitions, int elementsPerPartition, int remainder); @Override int numPartitions(); @Override int getPartition(Object key); }
BalancedPartitioner extends Partitioner { @Override public int getPartition(Object key) { int elementIdx = key.hashCode(); if (elementIdx <= (elementsPerPartition + 1) * remainder) { int outputPartition = elementIdx / (elementsPerPartition + 1); if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (1): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } else { int numValsInLargerPartitions = remainder * (elementsPerPartition + 1); int idxInSmallerPartitions = elementIdx - numValsInLargerPartitions; int smallPartitionIdx = idxInSmallerPartitions / elementsPerPartition; int outputPartition = remainder + smallPartitionIdx; if (outputPartition >= numPartitions) { outputPartition = getRandom().nextInt(numPartitions); log.trace("Random partition assigned (2): elementIdx={}, numPartitions={}, elementsPerPartition={}, remainder={}", elementIdx, numPartitions, elementsPerPartition, remainder); } return outputPartition; } } BalancedPartitioner(int numPartitions, int elementsPerPartition, int remainder); @Override int numPartitions(); @Override int getPartition(Object key);  }