You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by je...@apache.org on 2019/12/03 22:55:47 UTC
[incubator-pinot] 01/01: Add targz to SegmentIndexCreationImpl
This is an automated email from the ASF dual-hosted git repository.
jenniferdai pushed a commit to branch trypt
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 4419a3b86c19b4abc5127161eb3797cf57a66496
Author: Jennifer Dai <jd...@linkedin.com>
AuthorDate: Tue Dec 3 14:57:41 2019 -0800
Add targz to SegmentIndexCreationImpl
---
.../generator/SegmentGeneratorConfig.java | 10 ++++++++
.../impl/SegmentIndexCreationDriverImpl.java | 20 +++++++++-------
.../hadoop/job/mappers/SegmentCreationMapper.java | 27 ++++++++--------------
.../pinot/ingestion/common/JobConfigConstants.java | 3 +++
...mentBuildPushOfflineClusterIntegrationTest.java | 4 +++-
5 files changed, 38 insertions(+), 26 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/generator/SegmentGeneratorConfig.java b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/generator/SegmentGeneratorConfig.java
index 1d42001..4e57c6f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/generator/SegmentGeneratorConfig.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/indexsegment/generator/SegmentGeneratorConfig.java
@@ -114,6 +114,7 @@ public class SegmentGeneratorConfig {
private boolean _onHeap = false;
private boolean _skipTimeValueCheck = false;
private boolean _nullHandlingEnabled = false;
+ private boolean _createTarGz = false;
public SegmentGeneratorConfig() {
}
@@ -164,6 +165,7 @@ public class SegmentGeneratorConfig {
_recordReaderPath = config._recordReaderPath;
_skipTimeValueCheck = config._skipTimeValueCheck;
_nullHandlingEnabled = config._nullHandlingEnabled;
+ _createTarGz = config._createTarGz;
}
/**
@@ -352,6 +354,14 @@ public class SegmentGeneratorConfig {
_recordReaderPath = recordReaderPath;
}
+ public boolean isCreateTarGz() {
+ return _createTarGz;
+ }
+
+ public void setCreateTarGz(boolean createTarGz) {
+ _createTarGz = createTarGz;
+ }
+
public String getOutDir() {
return _outDir;
}
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/creator/impl/SegmentIndexCreationDriverImpl.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/creator/impl/SegmentIndexCreationDriverImpl.java
index 3941f66..1fdaa7e 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/creator/impl/SegmentIndexCreationDriverImpl.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/creator/impl/SegmentIndexCreationDriverImpl.java
@@ -32,12 +32,8 @@ import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import org.apache.commons.io.FileUtils;
-import org.apache.pinot.spi.data.FieldSpec;
-import org.apache.pinot.spi.data.MetricFieldSpec;
-import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.common.data.StarTreeIndexSpec;
-import org.apache.pinot.spi.data.readers.GenericRow;
-import org.apache.pinot.spi.data.readers.RecordReader;
+import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.core.data.readers.RecordReaderFactory;
import org.apache.pinot.core.data.recordtransformer.CompositeTransformer;
import org.apache.pinot.core.data.recordtransformer.RecordTransformer;
@@ -65,6 +61,11 @@ import org.apache.pinot.core.startree.hll.HllUtil;
import org.apache.pinot.core.startree.v2.builder.MultipleTreesBuilder;
import org.apache.pinot.core.startree.v2.builder.StarTreeV2BuilderConfig;
import org.apache.pinot.core.util.CrcUtils;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.apache.pinot.spi.data.MetricFieldSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.data.readers.RecordReader;
import org.apache.pinot.startree.hll.HllConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -388,9 +389,6 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
// Move the temporary directory into its final location
FileUtils.moveDirectory(tempIndexDir, segmentOutputDir);
- // Delete the temporary directory
- FileUtils.deleteQuietly(tempIndexDir);
-
// Convert segment format if necessary
convertFormatIfNeeded(segmentOutputDir);
@@ -415,6 +413,12 @@ public class SegmentIndexCreationDriverImpl implements SegmentIndexCreationDrive
// Persist creation metadata to disk
persistCreationMeta(segmentOutputDir, crc, creationTime);
+ if (config.isCreateTarGz()) {
+ File tarGzPath = new File(outputDir, segmentName + TarGzCompressionUtils.TAR_GZ_FILE_EXTENSION);
+ TarGzCompressionUtils.createTarGzOfDirectory(segmentOutputDir.getAbsolutePath(), tarGzPath.getAbsolutePath());
+ FileUtils.deleteQuietly(segmentOutputDir);
+ }
+
LOGGER.info("Driver, record read time : {}", totalRecordReadTime);
LOGGER.info("Driver, stats collector time : {}", totalStatsCollectorTime);
LOGGER.info("Driver, indexing time : {}", totalIndexTime);
diff --git a/pinot-ingestion-jobs/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java b/pinot-ingestion-jobs/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
index 438fd9c..dfb02b6 100644
--- a/pinot-ingestion-jobs/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
+++ b/pinot-ingestion-jobs/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentCreationMapper.java
@@ -36,14 +36,9 @@ import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.pinot.common.config.SegmentsValidationAndRetentionConfig;
import org.apache.pinot.common.config.TableConfig;
-import org.apache.pinot.spi.data.Schema;
-import org.apache.pinot.spi.data.TimeFieldSpec;
import org.apache.pinot.common.utils.DataSize;
-import org.apache.pinot.spi.utils.JsonUtils;
-import org.apache.pinot.common.utils.TarGzCompressionUtils;
import org.apache.pinot.core.data.readers.CSVRecordReaderConfig;
import org.apache.pinot.core.data.readers.FileFormat;
-import org.apache.pinot.spi.data.readers.RecordReaderConfig;
import org.apache.pinot.core.data.readers.ThriftRecordReaderConfig;
import org.apache.pinot.core.indexsegment.generator.SegmentGeneratorConfig;
import org.apache.pinot.core.segment.creator.SegmentIndexCreationDriver;
@@ -52,6 +47,10 @@ import org.apache.pinot.core.segment.name.NormalizedDateSegmentNameGenerator;
import org.apache.pinot.core.segment.name.SegmentNameGenerator;
import org.apache.pinot.core.segment.name.SimpleSegmentNameGenerator;
import org.apache.pinot.ingestion.common.JobConfigConstants;
+import org.apache.pinot.spi.data.Schema;
+import org.apache.pinot.spi.data.TimeFieldSpec;
+import org.apache.pinot.spi.data.readers.RecordReaderConfig;
+import org.apache.pinot.spi.utils.JsonUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -73,6 +72,7 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
protected TableConfig _tableConfig;
protected String _recordReaderPath;
protected Path _readerConfigFile;
+ protected boolean _createTarGz;
// HDFS segment tar directory
protected Path _hdfsSegmentTarDir;
@@ -81,7 +81,6 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
protected File _localStagingDir;
protected File _localInputDir;
protected File _localSegmentDir;
- protected File _localSegmentTarDir;
/**
* Generate a relative output directory path when `useRelativePath` flag is on.
@@ -120,6 +119,7 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
_readerConfigFile = new Path(readerConfigFile);
}
_recordReaderPath = _jobConf.get(JobConfigConstants.RECORD_READER_PATH);
+ _createTarGz = Boolean.getBoolean(_jobConf.get(JobConfigConstants.CREATE_TAR_GZ));
// Set up segment name generator
String segmentNameGeneratorType =
@@ -153,18 +153,16 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
_localStagingDir = new File(LOCAL_TEMP_DIR);
_localInputDir = new File(_localStagingDir, "inputData");
_localSegmentDir = new File(_localStagingDir, "segments");
- _localSegmentTarDir = new File(_localStagingDir, JobConfigConstants.SEGMENT_TAR_DIR);
if (_localStagingDir.exists()) {
_logger.warn("Deleting existing file: {}", _localStagingDir);
FileUtils.forceDelete(_localStagingDir);
}
_logger
- .info("Making local temporary directories: {}, {}, {}", _localStagingDir, _localInputDir, _localSegmentTarDir);
+ .info("Making local temporary directories: {}, {}, {}", _localStagingDir, _localInputDir);
Preconditions.checkState(_localStagingDir.mkdirs());
Preconditions.checkState(_localInputDir.mkdir());
Preconditions.checkState(_localSegmentDir.mkdir());
- Preconditions.checkState(_localSegmentTarDir.mkdir());
_logger.info("*********************************************************************");
_logger.info("Raw Table Name: {}", _rawTableName);
@@ -176,7 +174,6 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
_logger.info("HDFS Segment Tar Directory: {}", _hdfsSegmentTarDir);
_logger.info("Local Staging Directory: {}", _localStagingDir);
_logger.info("Local Input Directory: {}", _localInputDir);
- _logger.info("Local Segment Tar Directory: {}", _localSegmentTarDir);
_logger.info("*********************************************************************");
}
@@ -233,6 +230,7 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
segmentGeneratorConfig.setReaderConfig(getReaderConfig(fileFormat));
}
segmentGeneratorConfig.setOnHeap(true);
+ segmentGeneratorConfig.setCreateTarGz(_createTarGz);
addAdditionalSegmentGeneratorConfigs(segmentGeneratorConfig, hdfsInputFile, sequenceId);
@@ -260,16 +258,11 @@ public class SegmentCreationMapper extends Mapper<LongWritable, Text, LongWritab
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.createTarGzOfDirectory(localSegmentDir.getPath(), localSegmentTarFile.getPath());
+ File localSegmentTarFile = new File(_localSegmentDir, segmentTarFileName);
- long uncompressedSegmentSize = FileUtils.sizeOf(localSegmentDir);
long compressedSegmentSize = FileUtils.sizeOf(localSegmentTarFile);
- _logger.info("Size for segment: {}, uncompressed: {}, compressed: {}", segmentName,
- DataSize.fromBytes(uncompressedSegmentSize), DataSize.fromBytes(compressedSegmentSize));
+ _logger.info("Size for segment: {}, compressed: {}", segmentName, DataSize.fromBytes(compressedSegmentSize));
Path hdfsSegmentTarFile = new Path(_hdfsSegmentTarDir, segmentTarFileName);
if (_useRelativePath) {
diff --git a/pinot-ingestion-jobs/pinot-ingestion-common/src/main/java/org/apache/pinot/ingestion/common/JobConfigConstants.java b/pinot-ingestion-jobs/pinot-ingestion-common/src/main/java/org/apache/pinot/ingestion/common/JobConfigConstants.java
index 9c9238e..9cb35b8 100644
--- a/pinot-ingestion-jobs/pinot-ingestion-common/src/main/java/org/apache/pinot/ingestion/common/JobConfigConstants.java
+++ b/pinot-ingestion-jobs/pinot-ingestion-common/src/main/java/org/apache/pinot/ingestion/common/JobConfigConstants.java
@@ -76,4 +76,7 @@ public class JobConfigConstants {
// Assign sequence ids to input files based at each local directory level
public static final String LOCAL_DIRECTORY_SEQUENCE_ID = "local.directory.sequence.id";
+
+ // Creates a tar gz file
+ public static final String CREATE_TAR_GZ = "create.tar.gz";
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HadoopSegmentBuildPushOfflineClusterIntegrationTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HadoopSegmentBuildPushOfflineClusterIntegrationTest.java
index 4c99f8c..73dc51c 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HadoopSegmentBuildPushOfflineClusterIntegrationTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/HadoopSegmentBuildPushOfflineClusterIntegrationTest.java
@@ -40,7 +40,6 @@ import org.apache.hadoop.fs.Path;
import org.apache.hadoop.mapreduce.v2.MiniMRYarnCluster;
import org.apache.pinot.common.config.ColumnPartitionConfig;
import org.apache.pinot.common.config.SegmentPartitionConfig;
-import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.core.data.partition.PartitionFunction;
import org.apache.pinot.core.data.partition.PartitionFunctionFactory;
import org.apache.pinot.core.indexsegment.generator.SegmentVersion;
@@ -48,6 +47,7 @@ import org.apache.pinot.hadoop.job.HadoopSegmentCreationJob;
import org.apache.pinot.hadoop.job.HadoopSegmentPreprocessingJob;
import org.apache.pinot.ingestion.common.JobConfigConstants;
import org.apache.pinot.ingestion.jobs.SegmentTarPushJob;
+import org.apache.pinot.spi.data.Schema;
import org.apache.pinot.util.TestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -174,6 +174,8 @@ public class HadoopSegmentBuildPushOfflineClusterIntegrationTest extends BaseClu
properties.setProperty(JobConfigConstants.PUSH_TO_HOSTS, getDefaultControllerConfiguration().getControllerHost());
properties.setProperty(JobConfigConstants.PUSH_TO_PORT, getDefaultControllerConfiguration().getControllerPort());
+ properties.setProperty(JobConfigConstants.CREATE_TAR_GZ, "true");
+
Properties preComputeProperties = new Properties();
preComputeProperties.putAll(properties);
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org