@Test public void testWithFsMetricsBranchesAndPartitions() throws IOException { File publishPaths[] = new File[] { Files.createTempDir(), Files.createTempDir(), }; try { List<File[]> branchPaths = Arrays.stream(publishPaths).map(branchPath -> new File[] { new File(branchPath, "1-2-3-4"), new File(branchPath, "5-6-7-8") }).collect(Collectors.toList()); branchPaths.forEach(partitionPaths -> Arrays.stream(partitionPaths).forEach(File::mkdir)); State s = buildDefaultState(2); String md = new GlobalMetadata().toJson(); s.removeProp(ConfigurationKeys.DATA_PUBLISHER_METADATA_OUTPUT_DIR); s.setProp(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_KEY + ".0", "true"); s.setProp(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_KEY + ".1", "true"); s.setProp(ConfigurationKeys.WRITER_METADATA_KEY + ".0", md); s.setProp(ConfigurationKeys.WRITER_METADATA_KEY + ".1", md); s.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR + ".0", publishPaths[0].getAbsolutePath()); s.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR + ".1", publishPaths[1].getAbsolutePath()); s.setProp(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, "false"); s.setProp(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR + ".0", "false"); s.setProp(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR + ".1", "false"); s.setProp(ConfigurationKeys.DATA_PUBLISHER_METADATA_OUTPUT_FILE, "metadata.json"); WorkUnitState wuState1 = new WorkUnitState(); FsWriterMetrics metrics1 = buildWriterMetrics("foo1.json", "1-2-3-4", 0, 10); FsWriterMetrics metrics2 = buildWriterMetrics("foo1.json", "5-6-7-8",10, 20); wuState1.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + ".0", "1-2-3-4"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + ".0", metrics1.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + ".0_0", "1-2-3-4"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + ".0_0", metrics2.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + ".0" + "_1", "5-6-7-8"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + ".0_1", metrics2.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_METADATA_KEY + ".0", md); addStateToWorkunit(s, wuState1); WorkUnitState wuState2 = new WorkUnitState(); FsWriterMetrics metrics3 = buildWriterMetrics("foo3.json", "1-2-3-4", 1, 1, 30); wuState2.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + ".1", "1-2-3-4"); wuState2.setProp(ConfigurationKeys.WRITER_METADATA_KEY + ".1", md); wuState2.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + ".1", metrics3.toJson()); addStateToWorkunit(s, wuState2); WorkUnitState wuState3 = new WorkUnitState(); FsWriterMetrics metrics4 = buildWriterMetrics("foo4.json", "5-6-7-8", 2, 55); wuState3.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + ".0", "5-6-7-8"); wuState3.setProp(ConfigurationKeys.WRITER_METADATA_KEY + ".0", md); wuState3.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + ".0", metrics4.toJson()); addStateToWorkunit(s, wuState3); BaseDataPublisher publisher = new BaseDataPublisher(s); publisher.publishMetadata(ImmutableList.of(wuState1, wuState2, wuState3)); checkMetadata(new File(branchPaths.get(0)[0], "metadata.json.0"), 1, 10, new FsWriterMetrics.FileInfo("foo1.json", 10)); checkMetadata(new File(branchPaths.get(0)[1], "metadata.json.0"), 2, 75, new FsWriterMetrics.FileInfo("foo1.json", 20), new FsWriterMetrics.FileInfo("foo4.json", 55)); checkMetadata(new File(branchPaths.get(1)[0], "metadata.json.1"), 1, 30, new FsWriterMetrics.FileInfo("foo3.json", 30)); } finally { Arrays.stream(publishPaths).forEach(dir -> { try { FileUtils.deleteDirectory(dir); } catch (IOException e) { throw new RuntimeException("IOError"); } }); } }
@Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); @Override void initialize(); @Override void close(); @Override void publishData(WorkUnitState state); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void publishMetadata(WorkUnitState state); }
BaseDataPublisher extends SingleTaskDataPublisher { @Override public void publishMetadata(Collection<? extends WorkUnitState> states) throws IOException { Set<String> partitions = new HashSet<>(); mergeMetadataAndCollectPartitionNames(states, partitions); partitions.removeIf(Objects::isNull); WorkUnitState anyState = states.iterator().next(); for (int branchId = 0; branchId < numBranches; branchId++) { String mdOutputPath = getMetadataOutputPathFromState(anyState, branchId); String userSpecifiedPath = getUserSpecifiedOutputPathFromState(anyState, branchId); if (partitions.isEmpty() || userSpecifiedPath != null) { publishMetadata(getMergedMetadataForPartitionAndBranch(null, branchId), branchId, getMetadataOutputFileForBranch(anyState, branchId)); } else { String metadataFilename = getMetadataFileNameForBranch(anyState, branchId); if (mdOutputPath == null || metadataFilename == null) { LOG.info("Metadata filename not set for branch " + String.valueOf(branchId) + ": not publishing metadata."); continue; } for (String partition : partitions) { publishMetadata(getMergedMetadataForPartitionAndBranch(partition, branchId), branchId, new Path(new Path(mdOutputPath, partition), metadataFilename)); } } } } BaseDataPublisher(State state); @Override void initialize(); @Override void close(); @Override void publishData(WorkUnitState state); @Override void publishData(Collection<? extends WorkUnitState> states); @Override void publishMetadata(Collection<? extends WorkUnitState> states); @Override void publishMetadata(WorkUnitState state);  }