@Test public void testExpandAndIntersectCubePartitions() { Map<String, List<Partition>> nodeAssignment = Maps.newHashMap(); nodeAssignment.put("cube1", Arrays.asList(new Partition(1), new Partition(3))); nodeAssignment.put("cube2", Arrays.asList(new Partition(2), new Partition(4))); nodeAssignment.put("cube3", Arrays.asList(new Partition(2))); nodeAssignment.put("cube5", Arrays.asList(new Partition(1), new Partition(3), new Partition(5), new Partition(7))); List<CubePartitionRoundRobinAssigner.CubePartition> cubePartitions = assigner .expandAndIntersectCubePartitions(nodeAssignment); System.out.println(cubePartitions); }
protected List<CubePartition> expandAndIntersectCubePartitions(Map<String, List<Partition>> nodeAssignment) { List<CubePartition> result = Lists.newArrayList(); Map<Partition, Set<String>> reverseMap = Maps.newTreeMap(); for (Map.Entry<String, List<Partition>> cubePartitionEntry : nodeAssignment.entrySet()) { String cubeName = cubePartitionEntry.getKey(); List<Partition> partitions = cubePartitionEntry.getValue(); for (Partition partition : partitions) { Set<String> cubes = reverseMap.get(partition); if (cubes == null) { cubes = Sets.newTreeSet(); reverseMap.put(partition, cubes); } cubes.add(cubeName); } } for (Map.Entry<Partition, Set<String>> partitionCubesEntry : reverseMap.entrySet()) { Partition partition = partitionCubesEntry.getKey(); Set<String> cubes = partitionCubesEntry.getValue(); for (String cube : cubes) { CubePartition cubePartition = new CubePartition(cube, partition); result.add(cubePartition); } } return result; }
CubePartitionRoundRobinAssigner implements Assigner { protected List<CubePartition> expandAndIntersectCubePartitions(Map<String, List<Partition>> nodeAssignment) { List<CubePartition> result = Lists.newArrayList(); Map<Partition, Set<String>> reverseMap = Maps.newTreeMap(); for (Map.Entry<String, List<Partition>> cubePartitionEntry : nodeAssignment.entrySet()) { String cubeName = cubePartitionEntry.getKey(); List<Partition> partitions = cubePartitionEntry.getValue(); for (Partition partition : partitions) { Set<String> cubes = reverseMap.get(partition); if (cubes == null) { cubes = Sets.newTreeSet(); reverseMap.put(partition, cubes); } cubes.add(cubeName); } } for (Map.Entry<Partition, Set<String>> partitionCubesEntry : reverseMap.entrySet()) { Partition partition = partitionCubesEntry.getKey(); Set<String> cubes = partitionCubesEntry.getValue(); for (String cube : cubes) { CubePartition cubePartition = new CubePartition(cube, partition); result.add(cubePartition); } } return result; } }
CubePartitionRoundRobinAssigner implements Assigner { protected List<CubePartition> expandAndIntersectCubePartitions(Map<String, List<Partition>> nodeAssignment) { List<CubePartition> result = Lists.newArrayList(); Map<Partition, Set<String>> reverseMap = Maps.newTreeMap(); for (Map.Entry<String, List<Partition>> cubePartitionEntry : nodeAssignment.entrySet()) { String cubeName = cubePartitionEntry.getKey(); List<Partition> partitions = cubePartitionEntry.getValue(); for (Partition partition : partitions) { Set<String> cubes = reverseMap.get(partition); if (cubes == null) { cubes = Sets.newTreeSet(); reverseMap.put(partition, cubes); } cubes.add(cubeName); } } for (Map.Entry<Partition, Set<String>> partitionCubesEntry : reverseMap.entrySet()) { Partition partition = partitionCubesEntry.getKey(); Set<String> cubes = partitionCubesEntry.getValue(); for (String cube : cubes) { CubePartition cubePartition = new CubePartition(cube, partition); result.add(cubePartition); } } return result; }  }
CubePartitionRoundRobinAssigner implements Assigner { protected List<CubePartition> expandAndIntersectCubePartitions(Map<String, List<Partition>> nodeAssignment) { List<CubePartition> result = Lists.newArrayList(); Map<Partition, Set<String>> reverseMap = Maps.newTreeMap(); for (Map.Entry<String, List<Partition>> cubePartitionEntry : nodeAssignment.entrySet()) { String cubeName = cubePartitionEntry.getKey(); List<Partition> partitions = cubePartitionEntry.getValue(); for (Partition partition : partitions) { Set<String> cubes = reverseMap.get(partition); if (cubes == null) { cubes = Sets.newTreeSet(); reverseMap.put(partition, cubes); } cubes.add(cubeName); } } for (Map.Entry<Partition, Set<String>> partitionCubesEntry : reverseMap.entrySet()) { Partition partition = partitionCubesEntry.getKey(); Set<String> cubes = partitionCubesEntry.getValue(); for (String cube : cubes) { CubePartition cubePartition = new CubePartition(cube, partition); result.add(cubePartition); } } return result; }  @Override Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets,
                                                                    List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments); @Override CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets,
                                 List<CubeAssignment> existingAssignments); void addToGroupAssignment(Map<String, List<Partition>> groupAssignment, String cubeName, Partition partition); }
CubePartitionRoundRobinAssigner implements Assigner { protected List<CubePartition> expandAndIntersectCubePartitions(Map<String, List<Partition>> nodeAssignment) { List<CubePartition> result = Lists.newArrayList(); Map<Partition, Set<String>> reverseMap = Maps.newTreeMap(); for (Map.Entry<String, List<Partition>> cubePartitionEntry : nodeAssignment.entrySet()) { String cubeName = cubePartitionEntry.getKey(); List<Partition> partitions = cubePartitionEntry.getValue(); for (Partition partition : partitions) { Set<String> cubes = reverseMap.get(partition); if (cubes == null) { cubes = Sets.newTreeSet(); reverseMap.put(partition, cubes); } cubes.add(cubeName); } } for (Map.Entry<Partition, Set<String>> partitionCubesEntry : reverseMap.entrySet()) { Partition partition = partitionCubesEntry.getKey(); Set<String> cubes = partitionCubesEntry.getValue(); for (String cube : cubes) { CubePartition cubePartition = new CubePartition(cube, partition); result.add(cubePartition); } } return result; }  @Override Map<Integer, Map<String, List<Partition>>> reBalancePlan(List<ReplicaSet> replicaSets,
                                                                    List<StreamingCubeInfo> cubes, List<CubeAssignment> existingAssignments); @Override CubeAssignment assign(StreamingCubeInfo cube, List<ReplicaSet> replicaSets,
                                 List<CubeAssignment> existingAssignments); void addToGroupAssignment(Map<String, List<Partition>> groupAssignment, String cubeName, Partition partition);  }