@Test public void replicationSatisfied() throws Exception { int numBlocks = 7; int replication = 3; createFileWithNoLocations(TEST_URI, numBlocks); LoadConfig config = new LoadConfig(TEST_URI, replication); Set<Pair<WorkerInfo, ArrayList<LoadTask>>> assignments = new LoadDefinition().selectExecutors(config, JOB_WORKERS, new SelectExecutorsContext(1, mJobServerContext)); int totalBlockLoads = 0; for (Pair<WorkerInfo, ArrayList<LoadTask>> assignment : assignments) { totalBlockLoads += assignment.getSecond().size(); } Assert.assertEquals(numBlocks * replication, totalBlockLoads); }
@Override public Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { Map<String, WorkerInfo> jobWorkersByAddress = jobWorkerInfoList.stream() .collect(Collectors.toMap(info -> info.getAddress().getHost(), info -> info)); List<String> missingJobWorkerHosts = new ArrayList<>(); List<BlockWorkerInfo> workers = new ArrayList<>(); for (BlockWorkerInfo worker : context.getFsContext().getCachedWorkers()) { if (jobWorkersByAddress.containsKey(worker.getNetAddress().getHost())) { workers.add(worker); } else { LOG.warn("Worker on host {} has no local job worker", worker.getNetAddress().getHost()); missingJobWorkerHosts.add(worker.getNetAddress().getHost()); } } Multimap<WorkerInfo, LoadTask> assignments = LinkedListMultimap.create(); AlluxioURI uri = new AlluxioURI(config.getFilePath()); for (FileBlockInfo blockInfo : context.getFileSystem().getStatus(uri).getFileBlockInfos()) { List<String> workersWithoutBlock = getWorkersWithoutBlock(workers, blockInfo); int neededReplicas = config.getReplication() - blockInfo.getBlockInfo().getLocations().size(); if (workersWithoutBlock.size() < neededReplicas) { String missingJobWorkersMessage = ""; if (!missingJobWorkerHosts.isEmpty()) { missingJobWorkersMessage = ". The following workers could not be used because they have " + "no local job workers: " + missingJobWorkerHosts; } throw new FailedPreconditionException(String.format( "Failed to find enough block workers to replicate to. Needed %s but only found %s. " + "Available workers without the block: %s" + missingJobWorkersMessage, neededReplicas, workersWithoutBlock.size(), workersWithoutBlock)); } Collections.shuffle(workersWithoutBlock); for (int i = 0; i < neededReplicas; i++) { String address = workersWithoutBlock.get(i); WorkerInfo jobWorker = jobWorkersByAddress.get(address); assignments.put(jobWorker, new LoadTask(blockInfo.getBlockInfo().getBlockId())); } } Set<Pair<WorkerInfo, ArrayList<LoadTask>>> result = Sets.newHashSet(); for (Map.Entry<WorkerInfo, Collection<LoadTask>> assignment : assignments.asMap().entrySet()) { Collection<LoadTask> loadTasks = assignment.getValue(); List<List<LoadTask>> partitionedTasks = CommonUtils.partition(Lists.newArrayList(loadTasks), JOBS_PER_WORKER); for (List<LoadTask> tasks : partitionedTasks) { if (!tasks.isEmpty()) { result.add(new Pair<>(assignment.getKey(), Lists.newArrayList(tasks))); } } } return result; }
LoadDefinition extends AbstractVoidPlanDefinition<LoadConfig, ArrayList<LoadTask>> { @Override public Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { Map<String, WorkerInfo> jobWorkersByAddress = jobWorkerInfoList.stream() .collect(Collectors.toMap(info -> info.getAddress().getHost(), info -> info)); List<String> missingJobWorkerHosts = new ArrayList<>(); List<BlockWorkerInfo> workers = new ArrayList<>(); for (BlockWorkerInfo worker : context.getFsContext().getCachedWorkers()) { if (jobWorkersByAddress.containsKey(worker.getNetAddress().getHost())) { workers.add(worker); } else { LOG.warn("Worker on host {} has no local job worker", worker.getNetAddress().getHost()); missingJobWorkerHosts.add(worker.getNetAddress().getHost()); } } Multimap<WorkerInfo, LoadTask> assignments = LinkedListMultimap.create(); AlluxioURI uri = new AlluxioURI(config.getFilePath()); for (FileBlockInfo blockInfo : context.getFileSystem().getStatus(uri).getFileBlockInfos()) { List<String> workersWithoutBlock = getWorkersWithoutBlock(workers, blockInfo); int neededReplicas = config.getReplication() - blockInfo.getBlockInfo().getLocations().size(); if (workersWithoutBlock.size() < neededReplicas) { String missingJobWorkersMessage = ""; if (!missingJobWorkerHosts.isEmpty()) { missingJobWorkersMessage = ". The following workers could not be used because they have " + "no local job workers: " + missingJobWorkerHosts; } throw new FailedPreconditionException(String.format( "Failed to find enough block workers to replicate to. Needed %s but only found %s. " + "Available workers without the block: %s" + missingJobWorkersMessage, neededReplicas, workersWithoutBlock.size(), workersWithoutBlock)); } Collections.shuffle(workersWithoutBlock); for (int i = 0; i < neededReplicas; i++) { String address = workersWithoutBlock.get(i); WorkerInfo jobWorker = jobWorkersByAddress.get(address); assignments.put(jobWorker, new LoadTask(blockInfo.getBlockInfo().getBlockId())); } } Set<Pair<WorkerInfo, ArrayList<LoadTask>>> result = Sets.newHashSet(); for (Map.Entry<WorkerInfo, Collection<LoadTask>> assignment : assignments.asMap().entrySet()) { Collection<LoadTask> loadTasks = assignment.getValue(); List<List<LoadTask>> partitionedTasks = CommonUtils.partition(Lists.newArrayList(loadTasks), JOBS_PER_WORKER); for (List<LoadTask> tasks : partitionedTasks) { if (!tasks.isEmpty()) { result.add(new Pair<>(assignment.getKey(), Lists.newArrayList(tasks))); } } } return result; } }
LoadDefinition extends AbstractVoidPlanDefinition<LoadConfig, ArrayList<LoadTask>> { @Override public Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { Map<String, WorkerInfo> jobWorkersByAddress = jobWorkerInfoList.stream() .collect(Collectors.toMap(info -> info.getAddress().getHost(), info -> info)); List<String> missingJobWorkerHosts = new ArrayList<>(); List<BlockWorkerInfo> workers = new ArrayList<>(); for (BlockWorkerInfo worker : context.getFsContext().getCachedWorkers()) { if (jobWorkersByAddress.containsKey(worker.getNetAddress().getHost())) { workers.add(worker); } else { LOG.warn("Worker on host {} has no local job worker", worker.getNetAddress().getHost()); missingJobWorkerHosts.add(worker.getNetAddress().getHost()); } } Multimap<WorkerInfo, LoadTask> assignments = LinkedListMultimap.create(); AlluxioURI uri = new AlluxioURI(config.getFilePath()); for (FileBlockInfo blockInfo : context.getFileSystem().getStatus(uri).getFileBlockInfos()) { List<String> workersWithoutBlock = getWorkersWithoutBlock(workers, blockInfo); int neededReplicas = config.getReplication() - blockInfo.getBlockInfo().getLocations().size(); if (workersWithoutBlock.size() < neededReplicas) { String missingJobWorkersMessage = ""; if (!missingJobWorkerHosts.isEmpty()) { missingJobWorkersMessage = ". The following workers could not be used because they have " + "no local job workers: " + missingJobWorkerHosts; } throw new FailedPreconditionException(String.format( "Failed to find enough block workers to replicate to. Needed %s but only found %s. " + "Available workers without the block: %s" + missingJobWorkersMessage, neededReplicas, workersWithoutBlock.size(), workersWithoutBlock)); } Collections.shuffle(workersWithoutBlock); for (int i = 0; i < neededReplicas; i++) { String address = workersWithoutBlock.get(i); WorkerInfo jobWorker = jobWorkersByAddress.get(address); assignments.put(jobWorker, new LoadTask(blockInfo.getBlockInfo().getBlockId())); } } Set<Pair<WorkerInfo, ArrayList<LoadTask>>> result = Sets.newHashSet(); for (Map.Entry<WorkerInfo, Collection<LoadTask>> assignment : assignments.asMap().entrySet()) { Collection<LoadTask> loadTasks = assignment.getValue(); List<List<LoadTask>> partitionedTasks = CommonUtils.partition(Lists.newArrayList(loadTasks), JOBS_PER_WORKER); for (List<LoadTask> tasks : partitionedTasks) { if (!tasks.isEmpty()) { result.add(new Pair<>(assignment.getKey(), Lists.newArrayList(tasks))); } } } return result; } LoadDefinition(); }
LoadDefinition extends AbstractVoidPlanDefinition<LoadConfig, ArrayList<LoadTask>> { @Override public Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { Map<String, WorkerInfo> jobWorkersByAddress = jobWorkerInfoList.stream() .collect(Collectors.toMap(info -> info.getAddress().getHost(), info -> info)); List<String> missingJobWorkerHosts = new ArrayList<>(); List<BlockWorkerInfo> workers = new ArrayList<>(); for (BlockWorkerInfo worker : context.getFsContext().getCachedWorkers()) { if (jobWorkersByAddress.containsKey(worker.getNetAddress().getHost())) { workers.add(worker); } else { LOG.warn("Worker on host {} has no local job worker", worker.getNetAddress().getHost()); missingJobWorkerHosts.add(worker.getNetAddress().getHost()); } } Multimap<WorkerInfo, LoadTask> assignments = LinkedListMultimap.create(); AlluxioURI uri = new AlluxioURI(config.getFilePath()); for (FileBlockInfo blockInfo : context.getFileSystem().getStatus(uri).getFileBlockInfos()) { List<String> workersWithoutBlock = getWorkersWithoutBlock(workers, blockInfo); int neededReplicas = config.getReplication() - blockInfo.getBlockInfo().getLocations().size(); if (workersWithoutBlock.size() < neededReplicas) { String missingJobWorkersMessage = ""; if (!missingJobWorkerHosts.isEmpty()) { missingJobWorkersMessage = ". The following workers could not be used because they have " + "no local job workers: " + missingJobWorkerHosts; } throw new FailedPreconditionException(String.format( "Failed to find enough block workers to replicate to. Needed %s but only found %s. " + "Available workers without the block: %s" + missingJobWorkersMessage, neededReplicas, workersWithoutBlock.size(), workersWithoutBlock)); } Collections.shuffle(workersWithoutBlock); for (int i = 0; i < neededReplicas; i++) { String address = workersWithoutBlock.get(i); WorkerInfo jobWorker = jobWorkersByAddress.get(address); assignments.put(jobWorker, new LoadTask(blockInfo.getBlockInfo().getBlockId())); } } Set<Pair<WorkerInfo, ArrayList<LoadTask>>> result = Sets.newHashSet(); for (Map.Entry<WorkerInfo, Collection<LoadTask>> assignment : assignments.asMap().entrySet()) { Collection<LoadTask> loadTasks = assignment.getValue(); List<List<LoadTask>> partitionedTasks = CommonUtils.partition(Lists.newArrayList(loadTasks), JOBS_PER_WORKER); for (List<LoadTask> tasks : partitionedTasks) { if (!tasks.isEmpty()) { result.add(new Pair<>(assignment.getKey(), Lists.newArrayList(tasks))); } } } return result; } LoadDefinition(); @Override Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config,
      List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context); @Override SerializableVoid runTask(LoadConfig config, ArrayList<LoadTask> tasks,
      RunTaskContext context); @Override Class<LoadConfig> getJobConfigClass(); }
LoadDefinition extends AbstractVoidPlanDefinition<LoadConfig, ArrayList<LoadTask>> { @Override public Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config, List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context) throws Exception { Map<String, WorkerInfo> jobWorkersByAddress = jobWorkerInfoList.stream() .collect(Collectors.toMap(info -> info.getAddress().getHost(), info -> info)); List<String> missingJobWorkerHosts = new ArrayList<>(); List<BlockWorkerInfo> workers = new ArrayList<>(); for (BlockWorkerInfo worker : context.getFsContext().getCachedWorkers()) { if (jobWorkersByAddress.containsKey(worker.getNetAddress().getHost())) { workers.add(worker); } else { LOG.warn("Worker on host {} has no local job worker", worker.getNetAddress().getHost()); missingJobWorkerHosts.add(worker.getNetAddress().getHost()); } } Multimap<WorkerInfo, LoadTask> assignments = LinkedListMultimap.create(); AlluxioURI uri = new AlluxioURI(config.getFilePath()); for (FileBlockInfo blockInfo : context.getFileSystem().getStatus(uri).getFileBlockInfos()) { List<String> workersWithoutBlock = getWorkersWithoutBlock(workers, blockInfo); int neededReplicas = config.getReplication() - blockInfo.getBlockInfo().getLocations().size(); if (workersWithoutBlock.size() < neededReplicas) { String missingJobWorkersMessage = ""; if (!missingJobWorkerHosts.isEmpty()) { missingJobWorkersMessage = ". The following workers could not be used because they have " + "no local job workers: " + missingJobWorkerHosts; } throw new FailedPreconditionException(String.format( "Failed to find enough block workers to replicate to. Needed %s but only found %s. " + "Available workers without the block: %s" + missingJobWorkersMessage, neededReplicas, workersWithoutBlock.size(), workersWithoutBlock)); } Collections.shuffle(workersWithoutBlock); for (int i = 0; i < neededReplicas; i++) { String address = workersWithoutBlock.get(i); WorkerInfo jobWorker = jobWorkersByAddress.get(address); assignments.put(jobWorker, new LoadTask(blockInfo.getBlockInfo().getBlockId())); } } Set<Pair<WorkerInfo, ArrayList<LoadTask>>> result = Sets.newHashSet(); for (Map.Entry<WorkerInfo, Collection<LoadTask>> assignment : assignments.asMap().entrySet()) { Collection<LoadTask> loadTasks = assignment.getValue(); List<List<LoadTask>> partitionedTasks = CommonUtils.partition(Lists.newArrayList(loadTasks), JOBS_PER_WORKER); for (List<LoadTask> tasks : partitionedTasks) { if (!tasks.isEmpty()) { result.add(new Pair<>(assignment.getKey(), Lists.newArrayList(tasks))); } } } return result; } LoadDefinition(); @Override Set<Pair<WorkerInfo, ArrayList<LoadTask>>> selectExecutors(LoadConfig config,
      List<WorkerInfo> jobWorkerInfoList, SelectExecutorsContext context); @Override SerializableVoid runTask(LoadConfig config, ArrayList<LoadTask> tasks,
      RunTaskContext context); @Override Class<LoadConfig> getJobConfigClass();  }