Skip to content

Improve progress reporter in SegmentCreationMapper #8129

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
snleee merged 1 commit into from
Feb 5, 2022
Merged

Improve progress reporter in SegmentCreationMapper #8129

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 29 additions & 28 deletions ...pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -275,6 +275,35 @@ protected void map(LongWritable key, Text value, Context context)
driver.init(segmentGeneratorConfig);
validateSchema(driver.getIngestionSchemaValidator());
driver.build();
String segmentName = driver.getSegmentName();
_logger.info("Finish creating segment: {} with sequence id: {}", segmentName, sequenceId);

File localSegmentDir = new File(_localSegmentDir, segmentName);
String segmentTarFileName = segmentName + JobConfigConstants.TAR_GZ_FILE_EXT;
File localSegmentTarFile = new File(_localSegmentTarDir, segmentTarFileName);
_logger.info("Tarring segment from: {} to: {}", localSegmentDir, localSegmentTarFile);
TarGzCompressionUtils.createTarGzFile(localSegmentDir, localSegmentTarFile);

long uncompressedSegmentSize = FileUtils.sizeOf(localSegmentDir);
long compressedSegmentSize = FileUtils.sizeOf(localSegmentTarFile);
_logger.info("Size for segment: {}, uncompressed: {}, compressed: {}", segmentName,
DataSizeUtils.fromBytes(uncompressedSegmentSize), DataSizeUtils.fromBytes(compressedSegmentSize));

Path hdfsSegmentTarFile = new Path(_hdfsSegmentTarDir, segmentTarFileName);
if (_useRelativePath) {
Path relativeOutputPath =
getRelativeOutputPath(new Path(_jobConf.get(JobConfigConstants.PATH_TO_INPUT)).toUri(),
hdfsInputFile.toUri(), _hdfsSegmentTarDir);
hdfsSegmentTarFile = new Path(relativeOutputPath, segmentTarFileName);
}
_logger.info("Copying segment tar file from: {} to: {}", localSegmentTarFile, hdfsSegmentTarFile);
FileSystem.get(hdfsSegmentTarFile.toUri(), _jobConf)
.copyFromLocalFile(true, true, new Path(localSegmentTarFile.getAbsolutePath()), hdfsSegmentTarFile);

context.write(new LongWritable(sequenceId), new Text(segmentTarFileName));
_logger
.info("Finish generating segment: {} with HDFS input file: {}, sequence id: {}", segmentName, hdfsInputFile,
sequenceId);
} catch (Exception e) {
_logger.error("Caught exception while creating segment with HDFS input file: {}, sequence id: {}", hdfsInputFile,
sequenceId, e);
Expand All @@ -286,34 +315,6 @@ protected void map(LongWritable key, Text value, Context context)
_logger.error("Failed to interrupt progress reporter thread: {}", progressReporterThread);
}
}
String segmentName = driver.getSegmentName();
_logger.info("Finish creating segment: {} with sequence id: {}", segmentName, sequenceId);

File localSegmentDir = new File(_localSegmentDir, segmentName);
String segmentTarFileName = segmentName + JobConfigConstants.TAR_GZ_FILE_EXT;
File localSegmentTarFile = new File(_localSegmentTarDir, segmentTarFileName);
_logger.info("Tarring segment from: {} to: {}", localSegmentDir, localSegmentTarFile);
TarGzCompressionUtils.createTarGzFile(localSegmentDir, localSegmentTarFile);

long uncompressedSegmentSize = FileUtils.sizeOf(localSegmentDir);
long compressedSegmentSize = FileUtils.sizeOf(localSegmentTarFile);
_logger.info("Size for segment: {}, uncompressed: {}, compressed: {}", segmentName,
DataSizeUtils.fromBytes(uncompressedSegmentSize), DataSizeUtils.fromBytes(compressedSegmentSize));

Path hdfsSegmentTarFile = new Path(_hdfsSegmentTarDir, segmentTarFileName);
if (_useRelativePath) {
Path relativeOutputPath =
getRelativeOutputPath(new Path(_jobConf.get(JobConfigConstants.PATH_TO_INPUT)).toUri(), hdfsInputFile.toUri(),
_hdfsSegmentTarDir);
hdfsSegmentTarFile = new Path(relativeOutputPath, segmentTarFileName);
}
_logger.info("Copying segment tar file from: {} to: {}", localSegmentTarFile, hdfsSegmentTarFile);
FileSystem.get(hdfsSegmentTarFile.toUri(), _jobConf)
.copyFromLocalFile(true, true, new Path(localSegmentTarFile.getAbsolutePath()), hdfsSegmentTarFile);

context.write(new LongWritable(sequenceId), new Text(segmentTarFileName));
_logger.info("Finish generating segment: {} with HDFS input file: {}, sequence id: {}", segmentName, hdfsInputFile,
sequenceId);
}

protected FileFormat getFileFormat(String fileName) {
Expand Down