class LocalityGroupPartitioner extends Partitioner<BulkIngestKey,Value> implements DelegatePartitioner {

    @Override
    public int getPartition(BulkIngestKey bKey, Value value, int reducers) {
        Text columnFamily = bKey.getKey().getColumnFamily();
        int index = colFams.indexOf(columnFamily);
        if (index == -1) {
            if (!hasSeenUnknownColFams) {
                log.warn("Unexpected column family: " + columnFamily);
                hasSeenUnknownColFams = true;
            }
            return colFams.size() % reducers; // spill over unrecognized partition
        }
        return index % reducers;
    }

}