kkrugler commented on a change in pull request #6506:
URL: https://github.com/apache/incubator-pinot/pull/6506#discussion_r568947541
##########
File path:
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-hadoop/src/main/java/org/apache/pinot/plugin/ingestion/batch/hadoop/HadoopSegmentGenerationJobRunner.java
##########
@@ -269,33 +300,80 @@ public void run()
throw new RuntimeException("Job failed: " + job);
}
- LOGGER.info("Trying to copy segment tars from staging directory: [{}] to
output directory [{}]", stagingDirURI,
+ LOGGER.info("Moving segment tars from staging directory [{}] to output
directory [{}]", stagingDirURI,
outputDirURI);
- outputDirFS.copy(new Path(stagingDir, SEGMENT_TAR_DIR).toUri(),
outputDirURI);
+ moveFiles(outputDirFS, new Path(stagingDir,
SEGMENT_TAR_SUBDIR_NAME).toUri(), outputDirURI, _spec.isOverwriteOutput());
} finally {
LOGGER.info("Trying to clean up staging directory: [{}]", stagingDirURI);
outputDirFS.delete(stagingDirURI, true);
}
}
+ /**
+ * Move all files from the <sourceDir> to the <destDir>, but don't delete
existing contents of destDir.
+ * If <overwrite> is true, and the source file exists in the destination
directory, then replace it, otherwise
+ * log a warning and continue. We assume that source and destination
directories are on the same filesystem,
+ * so that move() can be used.
+ *
+ * @param fs
+ * @param sourceDir
+ * @param destDir
+ * @param overwrite
+ * @throws IOException
+ * @throws URISyntaxException
+ */
+ private void moveFiles(PinotFS fs, URI sourceDir, URI destDir, boolean
overwrite) throws IOException, URISyntaxException {
+ for (String sourcePath : fs.listFiles(sourceDir, true)) {
+ URI sourceFileUri = SegmentGenerationUtils.getFileURI(sourcePath,
sourceDir);
+ String sourceFilename = FilenameUtils.getName(sourceFileUri.getPath());
+ URI destFileUri =
SegmentGenerationUtils.getRelativeOutputPath(sourceDir, sourceFileUri,
destDir).resolve(sourceFilename);
+
+ if (!overwrite && fs.exists(destFileUri)) {
+ LOGGER.warn("Can't overwrite existing output segment tar file: {}",
destFileUri);
+ } else {
+ fs.move(sourceFileUri, destFileUri, true);
+ }
+ }
+ }
+
/**
* Can be overridden to plug in custom mapper.
*/
protected Class<? extends Mapper<LongWritable, Text, LongWritable, Text>>
getMapperClass() {
return HadoopSegmentCreationMapper.class;
}
- protected void packPluginsToDistributedCache(Job job) {
+ /**
+ * We have to put our jar (which contains the mapper) in the distributed
cache and add it to the classpath,
+ * as otherwise it's not available (since the pinot-all jar - which is
bigger - is what we've set as our job jar).
+ *
+ * @param job
+ * @param outputDirFS
+ * @param stagingDirURI
+ * @throws Exception
+ */
+ protected void addMapperJarToDistributedCache(Job job, PinotFS outputDirFS,
URI stagingDirURI) throws Exception {
+ File ourJar = new
File(getClass().getProtectionDomain().getCodeSource().getLocation().toURI());
+ Path distributedCacheJar = new Path(stagingDirURI.toString(),
ourJar.getName());
+ outputDirFS.copyFromLocalFile(ourJar, distributedCacheJar.toUri());
+ job.addFileToClassPath(distributedCacheJar);
+ }
+
+ protected void packPluginsToDistributedCache(Job job, PinotFS outputDirFS,
URI stagingDirURI) {
File pluginsRootDir = new File(PluginManager.get().getPluginsRootDir());
if (pluginsRootDir.exists()) {
- File pluginsTarGzFile = new File(PINOT_PLUGINS_TAR_GZ);
try {
+ File pluginsTarGzFile = File.createTempFile("pinot-plugins-",
".tar.gz");
TarGzCompressionUtils.createTarGzFile(pluginsRootDir,
pluginsTarGzFile);
- } catch (IOException e) {
+
+ // Copy to staging directory
+ Path cachedPluginsTarball = new Path(stagingDirURI.toString(),
SegmentGenerationUtils.PINOT_PLUGINS_TAR_GZ);
+ outputDirFS.copyFromLocalFile(pluginsTarGzFile,
cachedPluginsTarball.toUri());
+ job.addCacheFile(cachedPluginsTarball.toUri());
+ } catch (Exception e) {
LOGGER.error("Failed to tar plugins directory", e);
Review comment:
Done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org
