@Test public void testWithFsMetricsAndPartitions() throws IOException { File publishPath = Files.createTempDir(); try { File part1 = new File(publishPath, "1-2-3-4"); part1.mkdir(); File part2 = new File(publishPath, "5-6-7-8"); part2.mkdir(); State s = buildDefaultState(1); String md = new GlobalMetadata().toJson(); s.removeProp(ConfigurationKeys.DATA_PUBLISHER_METADATA_OUTPUT_DIR); s.setProp(ConfigurationKeys.DATA_PUBLISH_WRITER_METADATA_KEY, "true"); s.setProp(ConfigurationKeys.WRITER_METADATA_KEY, md); s.setProp(ConfigurationKeys.DATA_PUBLISHER_FINAL_DIR, publishPath.getAbsolutePath()); s.setProp(ConfigurationKeys.DATA_PUBLISHER_APPEND_EXTRACT_TO_FINAL_DIR, "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, "1-2-3-4"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY, metrics1.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + "_0", "1-2-3-4"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + " _0", metrics2.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY + "_1", "5-6-7-8"); wuState1.setProp(FsDataWriter.FS_WRITER_METRICS_KEY + " _1", metrics2.toJson()); wuState1.setProp(ConfigurationKeys.WRITER_METADATA_KEY, md); addStateToWorkunit(s, wuState1); WorkUnitState wuState2 = new WorkUnitState(); FsWriterMetrics metrics3 = buildWriterMetrics("foo3.json", "1-2-3-4", 1, 30); wuState2.setProp(ConfigurationKeys.WRITER_PARTITION_PATH_KEY, "1-2-3-4"); wuState2.setProp(ConfigurationKeys.WRITER_METADATA_KEY, md); wuState2.setProp(FsDataWriter.FS_WRITER_METRICS_KEY, 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, "5-6-7-8"); wuState3.setProp(ConfigurationKeys.WRITER_METADATA_KEY, md); wuState3.setProp(FsDataWriter.FS_WRITER_METRICS_KEY, metrics4.toJson()); addStateToWorkunit(s, wuState3); BaseDataPublisher publisher = new BaseDataPublisher(s); publisher.publishMetadata(ImmutableList.of(wuState1, wuState2, wuState3)); checkMetadata(new File(part1, "metadata.json"), 2, 40, new FsWriterMetrics.FileInfo("foo3.json", 30), new FsWriterMetrics.FileInfo("foo1.json", 10)); checkMetadata(new File(part2, "metadata.json"), 2, 75, new FsWriterMetrics.FileInfo("foo1.json", 20), new FsWriterMetrics.FileInfo("foo4.json", 55)); } finally { FileUtils.deleteDirectory(publishPath); } }
@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);  }