Merge pull request #31 from fx19880617/master

Adding postfix config segment.name.postfix to hadoop segment creation job configs.
apache · Oct 29, 2015 · dd3b058 · dd3b058
2 parents 3372640 + 2fd4dca
commit dd3b058
Showing 1 changed file with 7 additions and 2 deletions.
diff --git a/...src/main/java/com/linkedin/pinot/hadoop/job/mapper/HadoopSegmentCreationMapReduceJob.java b/...src/main/java/com/linkedin/pinot/hadoop/job/mapper/HadoopSegmentCreationMapReduceJob.java
@@ -29,7 +29,6 @@
 import org.codehaus.jackson.map.ObjectMapper;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import org.stringtemplate.v4.compiler.STParser.list_return;
 
 import com.linkedin.pinot.common.data.Schema;
 import com.linkedin.pinot.common.utils.TarGzCompressionUtils;
@@ -49,6 +48,7 @@ public static class HadoopSegmentCreationMapper extends Mapper<LongWritable, Tex
     private String _inputFilePath;
     private String _outputPath;
     private String _tableName;
+    private String _postfix;
 
     private Path _currentHdfsWorkDir;
     private String _currentDiskWorkDir;
@@ -83,6 +83,7 @@ public void setup(Context context) throws IOException, InterruptedException {
 
       _outputPath = _properties.get("path.to.output");
       _tableName = _properties.get("segment.table.name");
+      _postfix = _properties.get("segment.name.postfix", null);
       if (_outputPath == null || _tableName == null) {
         throw new RuntimeException(
             "Missing configs: " +
@@ -160,7 +161,11 @@ private String createSegment(String dataFilePath, Schema schema, String seqId) t
 
       FileFormat fileFormat = getFileFormat(dataFilePath);
       segmentGeneratorConfig.setInputFileFormat(fileFormat);
-      segmentGeneratorConfig.setSegmentNamePostfix(seqId);
+      if (null != _postfix) {
+        segmentGeneratorConfig.setSegmentNamePostfix(String.format("%s-%s", _postfix, seqId));
+      } else {
+        segmentGeneratorConfig.setSegmentNamePostfix(seqId);
+      }
       segmentGeneratorConfig.setRecordeReaderConfig(getReaderConfig(fileFormat));
 
       segmentGeneratorConfig.setIndexOutputDir(_localDiskSegmentDirectory);