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/09/03 21:43:44 UTC
[incubator-pinot] branch cleanup created (now aff5441)
This is an automated email from the ASF dual-hosted git repository.
jenniferdai pushed a change to branch cleanup
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at aff5441 Adding preprocess test
This branch includes the following new commits:
new aff5441 Adding preprocess test
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: Adding preprocess test
Posted by je...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jenniferdai pushed a commit to branch cleanup
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit aff5441f3da4e6bb2695a460445a6cfa4cadc33c
Author: Jennifer Dai <jd...@linkedin.com>
AuthorDate: Wed Aug 28 14:35:01 2019 -0700
Adding preprocess test
---
.../name/NormalizedDateSegmentNameGenerator.java | 2 +-
.../pinot/hadoop/job/SegmentPreprocessingJob.java | 13 ++-
.../job/mappers/SegmentPreprocessingMapper.java | 29 +++----
.../pinot/integration/tests/ClusterTest.java | 8 +-
...mentBuildPushOfflineClusterIntegrationTest.java | 97 +++++++++++++---------
...ime_Performance_2014_100k_subset_nonulls.schema | 5 ++
6 files changed, 87 insertions(+), 67 deletions(-)
diff --git a/pinot-core/src/main/java/org/apache/pinot/core/segment/name/NormalizedDateSegmentNameGenerator.java b/pinot-core/src/main/java/org/apache/pinot/core/segment/name/NormalizedDateSegmentNameGenerator.java
index ff1ecc1..be5ac7f 100644
--- a/pinot-core/src/main/java/org/apache/pinot/core/segment/name/NormalizedDateSegmentNameGenerator.java
+++ b/pinot-core/src/main/java/org/apache/pinot/core/segment/name/NormalizedDateSegmentNameGenerator.java
@@ -44,7 +44,7 @@ public class NormalizedDateSegmentNameGenerator implements SegmentNameGenerator
private SimpleDateFormat _inputSDF;
public NormalizedDateSegmentNameGenerator(@Nullable String pushFrequency, @Nullable TimeUnit timeType, @Nullable String timeFormat) {
- new NormalizedDateSegmentNameGenerator("myTable", null, false, "APPEND", pushFrequency, timeType, timeFormat);
+ this("myTable", null, false, "APPEND", pushFrequency, timeType, timeFormat);
}
public NormalizedDateSegmentNameGenerator(String tableName, @Nullable String segmentNamePrefix,
diff --git a/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/SegmentPreprocessingJob.java b/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/SegmentPreprocessingJob.java
index 7838e10..cd7ca7f 100644
--- a/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/SegmentPreprocessingJob.java
+++ b/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/SegmentPreprocessingJob.java
@@ -104,7 +104,7 @@ public class SegmentPreprocessingJob extends BaseSegmentJob {
// get input/output paths.
_inputSegmentDir = Preconditions.checkNotNull(getPathFromProperty(JobConfigConstants.PATH_TO_INPUT));
- _preprocessedOutputDir = getPathFromProperty(_properties.getProperty(JobConfigConstants.PREPROCESS_PATH_TO_OUTPUT, getDefaultPreprocessPath()));
+ _preprocessedOutputDir = getPathFromProperty(JobConfigConstants.PREPROCESS_PATH_TO_OUTPUT);
_rawTableName = Preconditions.checkNotNull(_properties.getProperty(JobConfigConstants.SEGMENT_TABLE_NAME));
_pathToDependencyJar = getPathFromProperty(JobConfigConstants.PATH_TO_DEPS_JAR);
@@ -131,10 +131,6 @@ public class SegmentPreprocessingJob extends BaseSegmentJob {
_logger.info("*********************************************************************");
}
- private String getDefaultPreprocessPath() {
- return _inputSegmentDir + "/" + "preprocess";
- }
-
public void run()
throws Exception {
if (!_enablePreprocessing) {
@@ -148,11 +144,14 @@ public class SegmentPreprocessingJob extends BaseSegmentJob {
final List<Path> inputDataPaths = getDataFilePaths(_inputSegmentDir);
Preconditions.checkState(inputDataPaths.size() != 0, "No files in the input directory.");
+ JobPreparationHelper.setDirPermission(_fileSystem, _preprocessedOutputDir, _defaultPermissionsMask);
+
+ System.out.println("output: " + _preprocessedOutputDir);
+ System.out.println("output dir: " + _preprocessedOutputDir.toString());
if (_fileSystem.exists(_preprocessedOutputDir)) {
_logger.warn("Found the output folder {}, deleting it", _preprocessedOutputDir);
_fileSystem.delete(_preprocessedOutputDir, true);
}
- JobPreparationHelper.setDirPermission(_fileSystem, _preprocessedOutputDir, _defaultPermissionsMask);
setTableConfigAndSchema();
@@ -411,7 +410,7 @@ public class SegmentPreprocessingJob extends BaseSegmentJob {
job.getConfiguration().set(InternalConfigConstants.SEGMENT_TIME_FORMAT, _pinotTableSchema.getTimeFieldSpec().getOutgoingGranularitySpec().getTimeFormat());
job.getConfiguration().set(InternalConfigConstants.SEGMENT_PUSH_FREQUENCY, validationConfig.getSegmentPushFrequency());
try (DataFileStream<GenericRecord> dataStreamReader = getAvroReader(path)) {
- job.getConfiguration().set(InternalConfigConstants.TIME_COLUMN_VALUE, (String) dataStreamReader.next().get(timeColumnName));
+ job.getConfiguration().set(InternalConfigConstants.TIME_COLUMN_VALUE, Integer.toString((int) dataStreamReader.next().get(timeColumnName)));
}
}
}
diff --git a/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentPreprocessingMapper.java b/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentPreprocessingMapper.java
index c432e55..519357c 100644
--- a/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentPreprocessingMapper.java
+++ b/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentPreprocessingMapper.java
@@ -43,9 +43,11 @@ public class SegmentPreprocessingMapper extends Mapper<AvroKey<GenericRecord>, N
private Schema _outputKeySchema;
private Schema _outputSchema;
private boolean _enablePartitioning;
- private String _sampleNormalizedTimeColumnValue = null;
- private NormalizedDateSegmentNameGenerator _normalizedDateSegmentNameGenerator = null;
private boolean _isAppend = false;
+ private String _timeColumnValue;
+ private String _pushFrequency;
+ private String _timeFormat;
+ private TimeUnit _timeUnit;
@Override
public void setup(final Context context) {
@@ -58,15 +60,12 @@ public class SegmentPreprocessingMapper extends Mapper<AvroKey<GenericRecord>, N
_timeColumn = configuration.get(InternalConfigConstants.TIME_COLUMN_CONFIG);
// Get sample time column value
- String timeColumnValue = configuration.get(InternalConfigConstants.TIME_COLUMN_VALUE);
+ _timeColumnValue = configuration.get(InternalConfigConstants.TIME_COLUMN_VALUE);
- String pushFrequency = configuration.get(InternalConfigConstants.SEGMENT_PUSH_FREQUENCY);
+ _pushFrequency = configuration.get(InternalConfigConstants.SEGMENT_PUSH_FREQUENCY);
String timeType = configuration.get(InternalConfigConstants.SEGMENT_TIME_TYPE);
- String timeFormat = configuration.get(InternalConfigConstants.SEGMENT_TIME_FORMAT);
- TimeUnit timeUnit = TimeUnit.valueOf(timeType);
- // Normalize time column value
- _normalizedDateSegmentNameGenerator = new NormalizedDateSegmentNameGenerator(pushFrequency, timeUnit, timeFormat);
- _sampleNormalizedTimeColumnValue = _normalizedDateSegmentNameGenerator.getNormalizedDate(timeColumnValue);
+ _timeFormat = configuration.get(InternalConfigConstants.SEGMENT_TIME_FORMAT);
+ _timeUnit = TimeUnit.valueOf(timeType);
}
String sortedColumn = configuration.get(InternalConfigConstants.SORTED_COLUMN_CONFIG);
@@ -85,15 +84,17 @@ public class SegmentPreprocessingMapper extends Mapper<AvroKey<GenericRecord>, N
throws IOException, InterruptedException {
if (_isAppend) {
+ // Normalize time column value
+ NormalizedDateSegmentNameGenerator normalizedDateSegmentNameGenerator = new NormalizedDateSegmentNameGenerator(_pushFrequency, _timeUnit, _timeFormat);
+ String sampleNormalizedTimeColumnValue = normalizedDateSegmentNameGenerator.getNormalizedDate(_timeColumnValue);
// Normalize time column value and check against sample value
- String timeColumnValue = (String) record.datum().get(_timeColumn);
- String normalizedTimeColumnValue = _normalizedDateSegmentNameGenerator.getNormalizedDate(timeColumnValue);
+ String timeColumnValue = record.datum().get(_timeColumn).toString();
+ String normalizedTimeColumnValue = normalizedDateSegmentNameGenerator.getNormalizedDate(timeColumnValue);
- if (!normalizedTimeColumnValue.equals(_sampleNormalizedTimeColumnValue)) {
+ if (!normalizedTimeColumnValue.equals(sampleNormalizedTimeColumnValue)) {
// TODO: Create a custom exception and gracefully catch this exception outside, changing what the path to input
// into segment creation should be
- throw new IllegalArgumentException("Your segment spans multiple time units. Preprocess is not currently allowed for"
- + " these use cases");
+ LOGGER.error("This segment contains multiple time units. Sample is {}, current is {}", sampleNormalizedTimeColumnValue, normalizedTimeColumnValue);
}
}
diff --git a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
index 5de22ac..6c6ab81 100644
--- a/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
+++ b/pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/ClusterTest.java
@@ -317,7 +317,7 @@ public abstract class ClusterTest extends ControllerTest {
throws Exception {
TableConfig tableConfig =
getOfflineTableConfig(tableName, timeColumnName, timeType, brokerTenant, serverTenant, loadMode, segmentVersion,
- invertedIndexColumns, bloomFilterColumns, taskConfig, segmentPartitionConfig, sortedColumn);
+ invertedIndexColumns, bloomFilterColumns, taskConfig, segmentPartitionConfig, sortedColumn, "daily");
if (!isUsingNewConfigFormat()) {
sendPostRequest(_controllerRequestURLBuilder.forTableCreate(), tableConfig.toJsonConfigString());
@@ -333,7 +333,7 @@ public abstract class ClusterTest extends ControllerTest {
throws Exception {
TableConfig tableConfig =
getOfflineTableConfig(tableName, timeColumnName, timeType, brokerTenant, serverTenant, loadMode, segmentVersion,
- invertedIndexColumns, bloomFilterColumns, taskConfig, segmentPartitionConfig, sortedColumn);
+ invertedIndexColumns, bloomFilterColumns, taskConfig, segmentPartitionConfig, sortedColumn, "daily");
if (!isUsingNewConfigFormat()) {
sendPutRequest(_controllerRequestURLBuilder.forUpdateTableConfig(tableName), tableConfig.toJsonConfigString());
@@ -345,9 +345,9 @@ public abstract class ClusterTest extends ControllerTest {
private static TableConfig getOfflineTableConfig(String tableName, String timeColumnName, String timeType,
String brokerTenant, String serverTenant, String loadMode, SegmentVersion segmentVersion,
List<String> invertedIndexColumns, List<String> bloomFilterColumns, TableTaskConfig taskConfig,
- SegmentPartitionConfig segmentPartitionConfig, String sortedColumn) {
+ SegmentPartitionConfig segmentPartitionConfig, String sortedColumn, String segmentPushFrequency) {
return new TableConfig.Builder(Helix.TableType.OFFLINE).setTableName(tableName).setTimeColumnName(timeColumnName)
- .setTimeType(timeType).setNumReplicas(1).setBrokerTenant(brokerTenant).setServerTenant(serverTenant)
+ .setTimeType(timeType).setSegmentPushFrequency(segmentPushFrequency).setNumReplicas(1).setBrokerTenant(brokerTenant).setServerTenant(serverTenant)
.setLoadMode(loadMode).setSegmentVersion(segmentVersion.toString())
.setInvertedIndexColumns(invertedIndexColumns).setBloomFilterColumns(bloomFilterColumns)
.setTaskConfig(taskConfig).setSegmentPartitionConfig(segmentPartitionConfig).setSortedColumn(sortedColumn)
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 81707c8..ae0e3b8 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
@@ -20,15 +20,29 @@ package org.apache.pinot.integration.tests;
import java.io.File;
import java.io.IOException;
+import java.util.HashSet;
import java.util.List;
+import java.util.Map;
import java.util.Properties;
+import java.util.Set;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericData;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecord;
import org.apache.commons.io.FileUtils;
import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
+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.common.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;
import org.apache.pinot.hadoop.job.JobConfigConstants;
import org.apache.pinot.hadoop.job.SegmentCreationJob;
@@ -37,6 +51,7 @@ import org.apache.pinot.hadoop.job.SegmentTarPushJob;
import org.apache.pinot.util.TestUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
+import org.testng.Assert;
import org.testng.annotations.AfterClass;
import org.testng.annotations.BeforeClass;
import org.testng.annotations.Test;
@@ -170,8 +185,8 @@ public class HadoopSegmentBuildPushOfflineClusterIntegrationTest extends BaseClu
preComputeProperties.putAll(properties);
preComputeProperties.setProperty(JobConfigConstants.PATH_TO_INPUT, _avroDir.getPath());
+ preComputeProperties.setProperty(JobConfigConstants.ENABLE_PREPROCESSING, "true");
preComputeProperties.setProperty(JobConfigConstants.PREPROCESS_PATH_TO_OUTPUT, _preprocessingDir.getPath());
-// properties.setProperty(JobConfigConstants.PATH_TO_INPUT, _preprocessingDir.getPath());
// Run segment pre-processing job
SegmentPreprocessingJob segmentPreprocessingJob = new SegmentPreprocessingJob(preComputeProperties);
@@ -196,45 +211,45 @@ public class HadoopSegmentBuildPushOfflineClusterIntegrationTest extends BaseClu
}
private void verifyPreprocessingJob(Configuration preComputeConfig) throws IOException {
- // TODO: Uncomment once this job is released
-// // Fetch partitioning config and sorting config.
-// SegmentPartitionConfig segmentPartitionConfig = getSegmentPartitionConfig();
-// Map.Entry<String, ColumnPartitionConfig>
-// entry = segmentPartitionConfig.getColumnPartitionMap().entrySet().iterator().next();
-// String partitionColumn = entry.getKey();
-// String partitionFunctionString = entry.getValue().getFunctionName();
-// int numPartitions = entry.getValue().getNumPartitions();
-// PartitionFunction partitionFunction = PartitionFunctionFactory.getPartitionFunction(partitionFunctionString, numPartitions);
-// String sortedColumn = getSortedColumn();
-//
-// // Get output files.
-// FileSystem fileSystem = FileSystem.get(preComputeConfig);
-// FileStatus[] fileStatuses = fileSystem.listStatus(new Path(_preprocessingDir.getPath()));
-// Assert.assertEquals(fileStatuses.length, numPartitions, "Number of output file should be the same as the number of partitions.");
-//
-// Set<Integer> partitionIdSet = new HashSet<>();
-// Object previousObject;
-// for (FileStatus fileStatus : fileStatuses) {
-// Path avroFile = fileStatus.getPath();
-// DataFileStream<GenericRecord> dataFileStream = new DataFileStream<>(fileSystem.open(avroFile), new GenericDatumReader<>());
-//
-// // Reset hash set and previous object
-// partitionIdSet.clear();
-// previousObject = null;
-// while (dataFileStream.hasNext()) {
-// GenericRecord genericRecord = dataFileStream.next();
-// partitionIdSet.add(partitionFunction.getPartition(genericRecord.get(partitionColumn)));
-// Assert.assertEquals(partitionIdSet.size(), 1, "Partition Id should be the same within a file.");
-// org.apache.avro.Schema sortedColumnSchema = genericRecord.getSchema().getField(sortedColumn).schema();
-// Object currentObject = genericRecord.get(sortedColumn);
-// if (previousObject == null) {
-// previousObject = currentObject;
-// continue;
-// }
-// // The values of sorted column should be sorted in ascending order.
-// Assert.assertTrue(GenericData.get().compare(previousObject, currentObject, sortedColumnSchema) <= 0);
-// previousObject = currentObject;
-// }
-// }
+ // Fetch partitioning config and sorting config.
+ SegmentPartitionConfig segmentPartitionConfig = getSegmentPartitionConfig();
+ Map.Entry<String, ColumnPartitionConfig>
+ entry = segmentPartitionConfig.getColumnPartitionMap().entrySet().iterator().next();
+ String partitionColumn = entry.getKey();
+ String partitionFunctionString = entry.getValue().getFunctionName();
+ int numPartitions = entry.getValue().getNumPartitions();
+ PartitionFunction
+ partitionFunction = PartitionFunctionFactory.getPartitionFunction(partitionFunctionString, numPartitions);
+ String sortedColumn = getSortedColumn();
+
+ // Get output files.
+ FileSystem fileSystem = FileSystem.get(preComputeConfig);
+ FileStatus[] fileStatuses = fileSystem.listStatus(new Path(_preprocessingDir.getPath()));
+ Assert.assertEquals(fileStatuses.length, numPartitions, "Number of output file should be the same as the number of partitions.");
+
+ Set<Integer> partitionIdSet = new HashSet<>();
+ Object previousObject;
+ for (FileStatus fileStatus : fileStatuses) {
+ Path avroFile = fileStatus.getPath();
+ DataFileStream<GenericRecord> dataFileStream = new DataFileStream<>(fileSystem.open(avroFile), new GenericDatumReader<>());
+
+ // Reset hash set and previous object
+ partitionIdSet.clear();
+ previousObject = null;
+ while (dataFileStream.hasNext()) {
+ GenericRecord genericRecord = dataFileStream.next();
+ partitionIdSet.add(partitionFunction.getPartition(genericRecord.get(partitionColumn)));
+ Assert.assertEquals(partitionIdSet.size(), 1, "Partition Id should be the same within a file.");
+ org.apache.avro.Schema sortedColumnSchema = genericRecord.getSchema().getField(sortedColumn).schema();
+ Object currentObject = genericRecord.get(sortedColumn);
+ if (previousObject == null) {
+ previousObject = currentObject;
+ continue;
+ }
+ // The values of sorted column should be sorted in ascending order.
+ Assert.assertTrue(GenericData.get().compare(previousObject, currentObject, sortedColumnSchema) <= 0);
+ previousObject = currentObject;
+ }
+ }
}
}
diff --git a/pinot-integration-tests/src/test/resources/On_Time_On_Time_Performance_2014_100k_subset_nonulls.schema b/pinot-integration-tests/src/test/resources/On_Time_On_Time_Performance_2014_100k_subset_nonulls.schema
index 7dff51a..ef2ab0e 100644
--- a/pinot-integration-tests/src/test/resources/On_Time_On_Time_Performance_2014_100k_subset_nonulls.schema
+++ b/pinot-integration-tests/src/test/resources/On_Time_On_Time_Performance_2014_100k_subset_nonulls.schema
@@ -330,6 +330,11 @@
"name": "DaysSinceEpoch",
"dataType": "INT",
"timeType": "DAYS"
+ },
+ "outgoingGranularitySpec": {
+ "name": "DaysSinceEpoch",
+ "dataType": "INT",
+ "timeType": "DAYS",
}
}
}
\ No newline at end of file
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org