mcvsubbu commented on a change in pull request #7062:
URL: https://github.com/apache/incubator-pinot/pull/7062#discussion_r655756476
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
##########
@@ -215,123 +158,84 @@ private void fetchPartitioningConfig() {
_partitionFunction =
segmentPartitionConfig.getFunctionName(_partitionColumn);
}
} else {
- _logger.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
+ LOGGER.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
}
}
private void fetchSortingConfig() {
- // Fetch sorting info from table config.
+ if (!_preprocessingOperations.contains("sort")) {
+ LOGGER.info("Sorting is disabled.");
+ return;
+ }
+ // Fetch sorting info from table config first.
+ List<String> sortingColumns = new ArrayList<>();
+ List<FieldConfig> fieldConfigs = _tableConfig.getFieldConfigList();
+ if (fieldConfigs != null && !fieldConfigs.isEmpty()) {
+ for (FieldConfig fieldConfig : fieldConfigs) {
+ if (fieldConfig.getIndexType() == FieldConfig.IndexType.SORTED) {
+ sortingColumns.add(fieldConfig.getName());
+ }
+ }
+ }
+ if (!sortingColumns.isEmpty()) {
+ Preconditions.checkArgument(sortingColumns.size() == 1, "There should be
at most 1 sorted column in the table.");
+ _sortingColumn = sortingColumns.get(0);
+ return;
+ }
+
+ // There is no sorted column specified in field configs, try to find
sorted column from indexing config.
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
List<String> sortedColumns = indexingConfig.getSortedColumn();
if (sortedColumns != null) {
Preconditions.checkArgument(sortedColumns.size() <= 1, "There should be
at most 1 sorted column in the table.");
if (sortedColumns.size() == 1) {
- _sortedColumn = sortedColumns.get(0);
+ _sortingColumn = sortedColumns.get(0);
+ FieldSpec fieldSpec =
_pinotTableSchema.getFieldSpecFor(_sortingColumn);
+ Preconditions.checkState(fieldSpec != null, "Failed to find sorting
column: {} in the schema", _sortingColumn);
+ Preconditions
+ .checkState(fieldSpec.isSingleValueField(), "Cannot sort on
multi-value column: %s", _sortingColumn);
+ _sortingColumnType = fieldSpec.getDataType();
+ Preconditions.checkState(_sortingColumnType !=
FieldSpec.DataType.BYTES, "Cannot sort on BYTES column: %s",
Review comment:
Now we have a whole bunch of column types: JSON, STRUCT, MAP, LIST.
Maybe you should add a method in FieldSpec.DataType like `canBeASortedColumn()`
and just invoke that?
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/preprocess/AvroDataPreprocessingHelper.java
##########
@@ -0,0 +1,155 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.hadoop.job.preprocess;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.List;
+import java.util.zip.GZIPInputStream;
+import org.apache.avro.Schema;
+import org.apache.avro.file.DataFileStream;
+import org.apache.avro.generic.GenericDatumReader;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.mapred.AvroKey;
+import org.apache.avro.mapreduce.AvroJob;
+import org.apache.avro.mapreduce.AvroKeyInputFormat;
+import org.apache.avro.mapreduce.AvroKeyOutputFormat;
+import org.apache.avro.mapreduce.AvroMultipleOutputs;
+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.io.NullWritable;
+import org.apache.hadoop.mapreduce.Job;
+import org.apache.hadoop.mapreduce.Partitioner;
+import org.apache.hadoop.mapreduce.lib.output.LazyOutputFormat;
+import org.apache.pinot.hadoop.job.mappers.AvroDataPreprocessingMapper;
+import
org.apache.pinot.hadoop.job.partitioners.AvroDataPreprocessingPartitioner;
+import org.apache.pinot.hadoop.job.reducers.AvroDataPreprocessingReducer;
+import org.apache.pinot.hadoop.utils.preprocess.HadoopUtils;
+import org.apache.pinot.segment.spi.partition.PartitionFunctionFactory;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class AvroDataPreprocessingHelper extends DataPreprocessingHelper {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(AvroDataPreprocessingHelper.class);
+
+ public AvroDataPreprocessingHelper(List<Path> inputDataPaths, Path
outputPath) {
+ super(inputDataPaths, outputPath);
+ }
+
+ @Override
+ public Class<? extends Partitioner> getPartitioner() {
+ return AvroDataPreprocessingPartitioner.class;
+ }
+
+ @Override
+ public void setUpMapperReducerConfigs(Job job)
+ throws IOException {
+ Schema avroSchema = getAvroSchema(_sampleRawDataPath);
+ LOGGER.info("Avro schema is: {}", avroSchema.toString(true));
+ validateConfigsAgainstSchema(avroSchema);
+
+ job.setInputFormatClass(AvroKeyInputFormat.class);
+ job.setMapperClass(AvroDataPreprocessingMapper.class);
+
+ job.setReducerClass(AvroDataPreprocessingReducer.class);
+ AvroMultipleOutputs.addNamedOutput(job, "avro", AvroKeyOutputFormat.class,
avroSchema);
Review comment:
Can you use `RawDataFormat.AVRO.name()` here?
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
##########
@@ -199,11 +122,31 @@ public void run()
throw new RuntimeException("Job failed : " + job);
}
- _logger.info("Finished pre-processing job in {}ms",
(System.currentTimeMillis() - startTime));
+ LOGGER.info("Finished pre-processing job in {}ms",
(System.currentTimeMillis() - startTime));
+ }
+
+ private void fetchPreProcessingOperations() {
+ _preprocessingOperations = new HashSet<>();
+ TableCustomConfig customConfig = _tableConfig.getCustomConfig();
+ if (customConfig != null) {
+ Map<String, String> customConfigMap = customConfig.getCustomConfigs();
+ if (customConfigMap != null && !customConfigMap.isEmpty()) {
+ String preprocessingOperationsString =
+
customConfigMap.getOrDefault(InternalConfigConstants.PREPROCESS_OPERATIONS, "");
+ String[] preprocessingOpsArray =
preprocessingOperationsString.split(",");
Review comment:
May be a good idea to parse and split each one of them into an enum, and
use enums later? Think about it.
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/HadoopSegmentPreprocessingJob.java
##########
@@ -215,123 +158,84 @@ private void fetchPartitioningConfig() {
_partitionFunction =
segmentPartitionConfig.getFunctionName(_partitionColumn);
}
} else {
- _logger.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
+ LOGGER.info("Segment partition config is null for table: {}",
_tableConfig.getTableName());
}
}
private void fetchSortingConfig() {
- // Fetch sorting info from table config.
+ if (!_preprocessingOperations.contains("sort")) {
+ LOGGER.info("Sorting is disabled.");
+ return;
+ }
+ // Fetch sorting info from table config first.
+ List<String> sortingColumns = new ArrayList<>();
+ List<FieldConfig> fieldConfigs = _tableConfig.getFieldConfigList();
+ if (fieldConfigs != null && !fieldConfigs.isEmpty()) {
+ for (FieldConfig fieldConfig : fieldConfigs) {
+ if (fieldConfig.getIndexType() == FieldConfig.IndexType.SORTED) {
+ sortingColumns.add(fieldConfig.getName());
+ }
+ }
+ }
+ if (!sortingColumns.isEmpty()) {
+ Preconditions.checkArgument(sortingColumns.size() == 1, "There should be
at most 1 sorted column in the table.");
+ _sortingColumn = sortingColumns.get(0);
+ return;
+ }
+
+ // There is no sorted column specified in field configs, try to find
sorted column from indexing config.
IndexingConfig indexingConfig = _tableConfig.getIndexingConfig();
List<String> sortedColumns = indexingConfig.getSortedColumn();
if (sortedColumns != null) {
Preconditions.checkArgument(sortedColumns.size() <= 1, "There should be
at most 1 sorted column in the table.");
if (sortedColumns.size() == 1) {
- _sortedColumn = sortedColumns.get(0);
+ _sortingColumn = sortedColumns.get(0);
+ FieldSpec fieldSpec =
_pinotTableSchema.getFieldSpecFor(_sortingColumn);
+ Preconditions.checkState(fieldSpec != null, "Failed to find sorting
column: {} in the schema", _sortingColumn);
+ Preconditions
+ .checkState(fieldSpec.isSingleValueField(), "Cannot sort on
multi-value column: %s", _sortingColumn);
+ _sortingColumnType = fieldSpec.getDataType();
+ Preconditions.checkState(_sortingColumnType !=
FieldSpec.DataType.BYTES, "Cannot sort on BYTES column: %s",
+ _sortingColumn);
+ LOGGER.info("Sorting the data with column: {} of type: {}",
_sortingColumn, _sortingColumnType);
}
}
}
private void fetchResizingConfig() {
+ if (!_preprocessingOperations.contains("resize")) {
+ LOGGER.info("Resizing is disabled.");
+ return;
+ }
TableCustomConfig tableCustomConfig = _tableConfig.getCustomConfig();
if (tableCustomConfig == null) {
_numOutputFiles = 0;
return;
}
Map<String, String> customConfigsMap =
tableCustomConfig.getCustomConfigs();
- if (customConfigsMap != null &&
customConfigsMap.containsKey(InternalConfigConstants.PREPROCESS_NUM_FILES)) {
- _numOutputFiles =
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESS_NUM_FILES));
+ if (customConfigsMap != null &&
customConfigsMap.containsKey(InternalConfigConstants.PREPROCESSING_NUM_REDUCERS))
{
+ _numOutputFiles =
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESSING_NUM_REDUCERS));
Preconditions.checkState(_numOutputFiles > 0, String
- .format("The value of %s should be positive! Current value: %s",
InternalConfigConstants.PREPROCESS_NUM_FILES,
- _numOutputFiles));
+ .format("The value of %s should be positive! Current value: %s",
+ InternalConfigConstants.PREPROCESSING_NUM_REDUCERS,
_numOutputFiles));
} else {
_numOutputFiles = 0;
}
- }
- private void setMaxNumRecordsConfigIfSpecified(Job job) {
- TableCustomConfig tableCustomConfig = _tableConfig.getCustomConfig();
- if (tableCustomConfig == null) {
- return;
- }
- Map<String, String> customConfigsMap =
tableCustomConfig.getCustomConfigs();
if (customConfigsMap != null && customConfigsMap
- .containsKey(InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE)) {
+
.containsKey(InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE)) {
int maxNumRecords =
-
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE));
+
Integer.parseInt(customConfigsMap.get(InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE));
Preconditions.checkArgument(maxNumRecords > 0,
- "The value of " +
InternalConfigConstants.PARTITION_MAX_RECORDS_PER_FILE
+ "The value of " +
InternalConfigConstants.PREPROCESSING_MAX_NUM_RECORDS_PER_FILE
Review comment:
Should we have an in-built max number of files as well? Some check to
make sure that we are not getting a million 1-record file? Can be a TODO
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/OrcDataPreprocessingMapper.java
##########
@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pinot.hadoop.job.mappers;
+
+import com.google.common.base.Preconditions;
+import java.io.IOException;
+import java.util.List;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.NullWritable;
+import org.apache.hadoop.io.WritableComparable;
+import org.apache.hadoop.mapreduce.Mapper;
+import org.apache.orc.mapred.OrcStruct;
+import org.apache.orc.mapred.OrcValue;
+import org.apache.pinot.hadoop.job.InternalConfigConstants;
+import org.apache.pinot.hadoop.utils.preprocess.DataPreprocessingUtils;
+import org.apache.pinot.hadoop.utils.preprocess.OrcUtils;
+import org.apache.pinot.spi.data.FieldSpec;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+public class OrcDataPreprocessingMapper extends Mapper<NullWritable,
OrcStruct, WritableComparable, OrcValue> {
+ private static final Logger LOGGER =
LoggerFactory.getLogger(OrcDataPreprocessingMapper.class);
+
+ private final OrcValue _valueWrapper = new OrcValue();
+ private String _sortingColumn = null;
+ private FieldSpec.DataType _sortingColumnType = null;
+ private int _sortingColumnId = -1;
+
+ @Override
+ public void setup(Context context) {
+ Configuration configuration = context.getConfiguration();
+ String sortingColumnConfig =
configuration.get(InternalConfigConstants.SORTING_COLUMN_CONFIG);
+ if (sortingColumnConfig != null) {
+ _sortingColumn = sortingColumnConfig;
+ _sortingColumnType =
FieldSpec.DataType.valueOf(configuration.get(InternalConfigConstants.SORTING_COLUMN_TYPE));
+ LOGGER.info("Initialized OrcDataPreprocessingMapper with sortingColumn:
{} of type: {}", _sortingColumn,
+ _sortingColumnType);
+ } else {
+ LOGGER.info("Initialized OrcDataPreprocessingMapper without sorting
column");
+ }
+ }
+
+ @Override
+ public void map(NullWritable key, OrcStruct value, Context context)
+ throws IOException, InterruptedException {
+ _valueWrapper.value = value;
+ if (_sortingColumn != null) {
+ if (_sortingColumnId == -1) {
+ List<String> fieldNames = value.getSchema().getFieldNames();
+ _sortingColumnId = fieldNames.indexOf(_sortingColumn);
+ Preconditions.checkState(_sortingColumnId != -1, "Failed to find
sorting column: %s in the ORC fields: %s",
+ _sortingColumn, fieldNames);
+ LOGGER.info("Field id for sorting column: {} is: {}", _sortingColumn,
_sortingColumnId);
Review comment:
Will this log appear for each record processed?
##########
File path:
pinot-plugins/pinot-batch-ingestion/v0_deprecated/pinot-hadoop/src/main/java/org/apache/pinot/hadoop/job/mappers/SegmentPreprocessingMapper.java
##########
@@ -130,4 +131,26 @@ public void map(AvroKey<GenericRecord> record,
NullWritable value, final Context
throw e;
}
}
+
+ protected void logConfigurations() {
+ StringBuilder stringBuilder = new StringBuilder();
+ stringBuilder.append('{');
+ boolean firstEntry = true;
+ for (Map.Entry<String, String> entry : _jobConf) {
+ if (!firstEntry) {
+ stringBuilder.append(", ");
+ } else {
+ firstEntry = false;
+ }
+
+ stringBuilder.append(entry.getKey());
+ stringBuilder.append('=');
+ stringBuilder.append(entry.getValue());
+ }
+ stringBuilder.append('}');
+
+
LOGGER.info("*********************************************************************");
+ LOGGER.info("Job Configurations: {}", stringBuilder.toString());
Review comment:
Will this be one very long log line? Will be a good idea to add some
newlines in the stringbuilder.append above.
--
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
