You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by GitBox <gi...@apache.org> on 2020/09/23 00:10:34 UTC
[GitHub] [incubator-pinot] npawar opened a new pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
npawar opened a new pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050
https://github.com/apache/incubator-pinot/issues/5753
The minion task executor which receives
1) segments (from realtime tables)
2) a time window
And then creates Pinot segments using data from that time window.
Uses the SegmentProcessorFramework. Applies:
1. Time column transformations as configured in PinotTaskConfig
2. Partitioning as specified in table config
3. Data sorting as specified in table config
4. Aggregations across common dimension+time, as configured in the PinotTaskConfig
Next steps:
RealtimeToOfflineSegmentsTaskGenerator
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497125779
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -61,4 +61,17 @@ private MinionConstants() {
public static final String MERGE_TYPE_KEY = "mergeTypeKey";
public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
}
+
+ public static class RealtimeToOfflineSegmentsTask {
+ public static final String TASK_TYPE = "realtimeToOfflineSegmentsTask";
+ // window
Review comment:
I see. I will make the changes with my merge & roll-up task generator PR.
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r496245347
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -61,4 +61,17 @@ private MinionConstants() {
public static final String MERGE_TYPE_KEY = "mergeTypeKey";
public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
}
+
+ public static class RealtimeToOfflineSegmentsTask {
+ public static final String TASK_TYPE = "realtimeToOfflineSegmentsTask";
+ // window
+ public static final String WINDOW_START_MILLIS_KEY = "windowStartMillis";
+ public static final String WINDOW_END_MILLIS_KEY = "windowEndMillis";
+ // segment processing
+ public static final String TIME_COLUMN_TRANSFORM_FUNCTION_KEY = "timeColumnTransformFunction";
+ public static final String COLLECTOR_TYPE_KEY = "collectorType";
+ public static final String AGGREGATION_TYPE_KEY_SUFFIX = ".aggregationType";
+ public static final String NUM_RECORDS_PER_SEGMENT_KEY = "numRecordsPerSegment";
Review comment:
`maxNumRecordsPerSegment` makes more sense since this configuration is indeed the upper bound of the number of rows in the segment
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -61,4 +61,17 @@ private MinionConstants() {
public static final String MERGE_TYPE_KEY = "mergeTypeKey";
public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
}
+
+ public static class RealtimeToOfflineSegmentsTask {
+ public static final String TASK_TYPE = "realtimeToOfflineSegmentsTask";
+ // window
Review comment:
Let's move the segment processing framework related configs to a static separate class. These configs will also be used by other tasks.
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
Review comment:
+1
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
Review comment:
Does this mean that we will drop the late data once the data from a window get moved to offline?
e.g.
1. day1 gets moved to offline table
2. day1 data arrived late (this row shows up in the result because of realtime table will index and serve this row)
3. day2 gets moved to offline table (the above late data will be dropped by the filter)
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] npawar merged pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
npawar merged pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r494438754
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
Review comment:
Can you use Preconditions instead of assert? Some runtime environments turns assert off. Thanks.
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
Review comment:
Good to add a duration in this log.
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Cannot partition using more than 1 ColumnPartitionConfig for table: %s", tableNameWithType);
+ String partitionColumn = columnPartitionMap.keySet().iterator().next();
+ Preconditions.checkState(schemaColumns.contains(partitionColumn),
+ "Partition column: %s is not a physical column in the schema", partitionColumn);
+ return new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
+ .setColumnName(partitionColumn).setColumnPartitionConfig(columnPartitionMap.get(partitionColumn)).build();
+ }
+
+ /**
+ * Construct a {@link CollectorConfig} using configured collector configs and sorted columns from table config
+ */
+ private CollectorConfig getCollectorConfig(String collectorTypeStr, Map<String, String> aggregateConfigs,
+ Set<String> schemaColumns, List<String> sortedColumns) {
+ CollectorFactory.CollectorType collectorType = collectorTypeStr == null ? CollectorFactory.CollectorType.CONCAT
+ : CollectorFactory.CollectorType.valueOf(collectorTypeStr.toUpperCase());
+
+ Map<String, ValueAggregatorFactory.ValueAggregatorType> aggregatorTypeMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : aggregateConfigs.entrySet()) {
+ String column = entry.getKey();
+ Preconditions
+ .checkState(schemaColumns.contains(column), "Aggregate column: %s is not a physical column in the schema",
+ column);
+ aggregatorTypeMap.put(column, ValueAggregatorFactory.ValueAggregatorType.valueOf(entry.getValue().toUpperCase()));
+ }
+
+ if (sortedColumns != null) {
+ for (String column : sortedColumns) {
Review comment:
why this check, given that we are pulling from tableconfig (at least in this task). If you are checking, then you should perhaps also check that this sorted column is not same as one of the aggregation columns?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r494438754
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
Review comment:
Can you use Preconditions instead of assert? Some runtime environments turns assert off. Thanks.
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
Review comment:
Good to add a duration in this log.
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Cannot partition using more than 1 ColumnPartitionConfig for table: %s", tableNameWithType);
+ String partitionColumn = columnPartitionMap.keySet().iterator().next();
+ Preconditions.checkState(schemaColumns.contains(partitionColumn),
+ "Partition column: %s is not a physical column in the schema", partitionColumn);
+ return new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
+ .setColumnName(partitionColumn).setColumnPartitionConfig(columnPartitionMap.get(partitionColumn)).build();
+ }
+
+ /**
+ * Construct a {@link CollectorConfig} using configured collector configs and sorted columns from table config
+ */
+ private CollectorConfig getCollectorConfig(String collectorTypeStr, Map<String, String> aggregateConfigs,
+ Set<String> schemaColumns, List<String> sortedColumns) {
+ CollectorFactory.CollectorType collectorType = collectorTypeStr == null ? CollectorFactory.CollectorType.CONCAT
+ : CollectorFactory.CollectorType.valueOf(collectorTypeStr.toUpperCase());
+
+ Map<String, ValueAggregatorFactory.ValueAggregatorType> aggregatorTypeMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : aggregateConfigs.entrySet()) {
+ String column = entry.getKey();
+ Preconditions
+ .checkState(schemaColumns.contains(column), "Aggregate column: %s is not a physical column in the schema",
+ column);
+ aggregatorTypeMap.put(column, ValueAggregatorFactory.ValueAggregatorType.valueOf(entry.getValue().toUpperCase()));
+ }
+
+ if (sortedColumns != null) {
+ for (String column : sortedColumns) {
Review comment:
why this check, given that we are pulling from tableconfig (at least in this task). If you are checking, then you should perhaps also check that this sorted column is not same as one of the aggregation columns?
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
npawar commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497115105
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Cannot partition using more than 1 ColumnPartitionConfig for table: %s", tableNameWithType);
+ String partitionColumn = columnPartitionMap.keySet().iterator().next();
+ Preconditions.checkState(schemaColumns.contains(partitionColumn),
+ "Partition column: %s is not a physical column in the schema", partitionColumn);
+ return new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
+ .setColumnName(partitionColumn).setColumnPartitionConfig(columnPartitionMap.get(partitionColumn)).build();
+ }
+
+ /**
+ * Construct a {@link CollectorConfig} using configured collector configs and sorted columns from table config
+ */
+ private CollectorConfig getCollectorConfig(String collectorTypeStr, Map<String, String> aggregateConfigs,
+ Set<String> schemaColumns, List<String> sortedColumns) {
+ CollectorFactory.CollectorType collectorType = collectorTypeStr == null ? CollectorFactory.CollectorType.CONCAT
+ : CollectorFactory.CollectorType.valueOf(collectorTypeStr.toUpperCase());
+
+ Map<String, ValueAggregatorFactory.ValueAggregatorType> aggregatorTypeMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : aggregateConfigs.entrySet()) {
+ String column = entry.getKey();
+ Preconditions
+ .checkState(schemaColumns.contains(column), "Aggregate column: %s is not a physical column in the schema",
+ column);
+ aggregatorTypeMap.put(column, ValueAggregatorFactory.ValueAggregatorType.valueOf(entry.getValue().toUpperCase()));
+ }
+
+ if (sortedColumns != null) {
+ for (String column : sortedColumns) {
Review comment:
Recent table config validation efforts showed that we hadn't been checking for validity of column names used in indexing config. I want to prevent failing in the SegmentProcessorFramework as much as possible, hence the check. If this check is not done, then we'd see failure in reduce step, after a lot of wasted computation.
I don't think we need to check for sorted column not being a metric, as nothing is really stopping a user from setting it that way
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
npawar commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497693069
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Cannot partition using more than 1 ColumnPartitionConfig for table: %s", tableNameWithType);
+ String partitionColumn = columnPartitionMap.keySet().iterator().next();
+ Preconditions.checkState(schemaColumns.contains(partitionColumn),
+ "Partition column: %s is not a physical column in the schema", partitionColumn);
+ return new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
+ .setColumnName(partitionColumn).setColumnPartitionConfig(columnPartitionMap.get(partitionColumn)).build();
+ }
+
+ /**
+ * Construct a {@link CollectorConfig} using configured collector configs and sorted columns from table config
+ */
+ private CollectorConfig getCollectorConfig(String collectorTypeStr, Map<String, String> aggregateConfigs,
+ Set<String> schemaColumns, List<String> sortedColumns) {
+ CollectorFactory.CollectorType collectorType = collectorTypeStr == null ? CollectorFactory.CollectorType.CONCAT
+ : CollectorFactory.CollectorType.valueOf(collectorTypeStr.toUpperCase());
+
+ Map<String, ValueAggregatorFactory.ValueAggregatorType> aggregatorTypeMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : aggregateConfigs.entrySet()) {
+ String column = entry.getKey();
+ Preconditions
+ .checkState(schemaColumns.contains(column), "Aggregate column: %s is not a physical column in the schema",
+ column);
+ aggregatorTypeMap.put(column, ValueAggregatorFactory.ValueAggregatorType.valueOf(entry.getValue().toUpperCase()));
+ }
+
+ if (sortedColumns != null) {
+ for (String column : sortedColumns) {
Review comment:
that's a valid scenario imo
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
npawar commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497117811
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
Review comment:
yes this will happen. this has been called out in the design doc: https://docs.google.com/document/d/1-e_9aHQB4HXS38ONtofdxNvMsGmAoYfSnc2LP88MbIc/edit#heading=h.5lkm0pm6vp7o
We discussed that the only way to handle this is for user to set a sufficient "bufferTime".
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] npawar commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
npawar commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497118618
##########
File path: pinot-core/src/main/java/org/apache/pinot/core/common/MinionConstants.java
##########
@@ -61,4 +61,17 @@ private MinionConstants() {
public static final String MERGE_TYPE_KEY = "mergeTypeKey";
public static final String MERGED_SEGMENT_NAME_KEY = "mergedSegmentNameKey";
}
+
+ public static class RealtimeToOfflineSegmentsTask {
+ public static final String TASK_TYPE = "realtimeToOfflineSegmentsTask";
+ // window
Review comment:
Can we do that when we add these other tasks? It will read odd for this PR
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] mcvsubbu commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
mcvsubbu commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497145807
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,275 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ assert dateTimeFieldSpec != null;
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ LOGGER.info("Finished task: {} with configs: {}", taskType, configs);
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
+ Preconditions.checkState(columnPartitionMap.size() == 1,
+ "Cannot partition using more than 1 ColumnPartitionConfig for table: %s", tableNameWithType);
+ String partitionColumn = columnPartitionMap.keySet().iterator().next();
+ Preconditions.checkState(schemaColumns.contains(partitionColumn),
+ "Partition column: %s is not a physical column in the schema", partitionColumn);
+ return new PartitionerConfig.Builder().setPartitionerType(PartitionerFactory.PartitionerType.TABLE_PARTITION_CONFIG)
+ .setColumnName(partitionColumn).setColumnPartitionConfig(columnPartitionMap.get(partitionColumn)).build();
+ }
+
+ /**
+ * Construct a {@link CollectorConfig} using configured collector configs and sorted columns from table config
+ */
+ private CollectorConfig getCollectorConfig(String collectorTypeStr, Map<String, String> aggregateConfigs,
+ Set<String> schemaColumns, List<String> sortedColumns) {
+ CollectorFactory.CollectorType collectorType = collectorTypeStr == null ? CollectorFactory.CollectorType.CONCAT
+ : CollectorFactory.CollectorType.valueOf(collectorTypeStr.toUpperCase());
+
+ Map<String, ValueAggregatorFactory.ValueAggregatorType> aggregatorTypeMap = new HashMap<>();
+ for (Map.Entry<String, String> entry : aggregateConfigs.entrySet()) {
+ String column = entry.getKey();
+ Preconditions
+ .checkState(schemaColumns.contains(column), "Aggregate column: %s is not a physical column in the schema",
+ column);
+ aggregatorTypeMap.put(column, ValueAggregatorFactory.ValueAggregatorType.valueOf(entry.getValue().toUpperCase()));
+ }
+
+ if (sortedColumns != null) {
+ for (String column : sortedColumns) {
Review comment:
I meant if a column is present both as sortedColumn, but specified in `aggregateConfigs` above. Not sure if that case is handled correctly
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[GitHub] [incubator-pinot] snleee commented on a change in pull request #6050: Minion taskExecutor for RealtimeToOfflineSegments task
Posted by GitBox <gi...@apache.org>.
snleee commented on a change in pull request #6050:
URL: https://github.com/apache/incubator-pinot/pull/6050#discussion_r497653163
##########
File path: pinot-minion/src/main/java/org/apache/pinot/minion/executor/RealtimeToOfflineSegmentsTaskExecutor.java
##########
@@ -0,0 +1,279 @@
+/**
+ * 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.minion.executor;
+
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
+import java.io.File;
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import org.apache.commons.io.FileUtils;
+import org.apache.pinot.core.common.MinionConstants;
+import org.apache.pinot.core.minion.PinotTaskConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorConfig;
+import org.apache.pinot.core.segment.processing.collector.CollectorFactory;
+import org.apache.pinot.core.segment.processing.collector.ValueAggregatorFactory;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterConfig;
+import org.apache.pinot.core.segment.processing.filter.RecordFilterFactory;
+import org.apache.pinot.core.segment.processing.framework.SegmentConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorConfig;
+import org.apache.pinot.core.segment.processing.framework.SegmentProcessorFramework;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerConfig;
+import org.apache.pinot.core.segment.processing.partitioner.PartitionerFactory;
+import org.apache.pinot.core.segment.processing.transformer.RecordTransformerConfig;
+import org.apache.pinot.spi.config.table.ColumnPartitionConfig;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.DateTimeFieldSpec;
+import org.apache.pinot.spi.data.DateTimeFormatSpec;
+import org.apache.pinot.spi.data.Schema;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * A task to convert segments from a REALTIME table to segments for its corresponding OFFLINE table.
+ * The realtime segments could span across multiple time windows. This task extracts data and creates segments for a configured time range.
+ * The {@link SegmentProcessorFramework} is used for the segment conversion, which also does
+ * 1. time column rollup
+ * 2. time window extraction using filter function
+ * 3. partitioning using table config's segmentPartitioningConfig
+ * 4. aggregations and rollup
+ * 5. data sorting
+ */
+public class RealtimeToOfflineSegmentsTaskExecutor extends BaseMultipleSegmentsConversionExecutor {
+ private static final Logger LOGGER = LoggerFactory.getLogger(RealtimeToOfflineSegmentsTaskExecutor.class);
+ private static final String INPUT_SEGMENTS_DIR = "input_segments";
+ private static final String OUTPUT_SEGMENTS_DIR = "output_segments";
+
+ @Override
+ protected List<SegmentConversionResult> convert(PinotTaskConfig pinotTaskConfig, List<File> originalIndexDirs,
+ File workingDir)
+ throws Exception {
+ String taskType = pinotTaskConfig.getTaskType();
+ Map<String, String> configs = pinotTaskConfig.getConfigs();
+ LOGGER.info("Starting task: {} with configs: {}", taskType, configs);
+ long startMillis = System.currentTimeMillis();
+
+ String tableNameWithType = configs.get(MinionConstants.TABLE_NAME_KEY); // rawTableName_OFFLINE expected here
+ TableConfig tableConfig = getTableConfig(tableNameWithType);
+ Schema schema = getSchema(tableNameWithType);
+ Set<String> schemaColumns = schema.getPhysicalColumnNames();
+ String timeColumn = tableConfig.getValidationConfig().getTimeColumnName();
+ DateTimeFieldSpec dateTimeFieldSpec = schema.getSpecForTimeColumn(timeColumn);
+ Preconditions
+ .checkState(dateTimeFieldSpec != null, "No valid spec found for time column: %s in schema for table: %s",
+ timeColumn, tableNameWithType);
+
+ long windowStartMs =
+ Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_START_MILLIS_KEY));
+ long windowEndMs = Long.parseLong(configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.WINDOW_END_MILLIS_KEY));
+ String timeColumnTransformFunction =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.TIME_COLUMN_TRANSFORM_FUNCTION_KEY);
+ String collectorTypeStr = configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.COLLECTOR_TYPE_KEY);
+ Map<String, String> aggregatorConfigs = new HashMap<>();
+ for (Map.Entry<String, String> entry : configs.entrySet()) {
+ String key = entry.getKey();
+ if (key.endsWith(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)) {
+ String column = key.split(MinionConstants.RealtimeToOfflineSegmentsTask.AGGREGATION_TYPE_KEY_SUFFIX)[0];
+ aggregatorConfigs.put(column, entry.getValue());
+ }
+ }
+ String numRecordsPerSegment =
+ configs.get(MinionConstants.RealtimeToOfflineSegmentsTask.MAX_NUM_RECORDS_PER_SEGMENT_KEY);
+
+ SegmentProcessorConfig.Builder segmentProcessorConfigBuilder =
+ new SegmentProcessorConfig.Builder().setTableConfig(tableConfig).setSchema(schema);
+
+ // Time rollup using configured time transformation function
+ if (timeColumnTransformFunction != null) {
+ RecordTransformerConfig recordTransformerConfig =
+ getRecordTransformerConfigForTime(timeColumnTransformFunction, timeColumn);
+ segmentProcessorConfigBuilder.setRecordTransformerConfig(recordTransformerConfig);
+ }
+
+ // Filter function for extracting data between start and end time window
+ RecordFilterConfig recordFilterConfig =
+ getRecordFilterConfigForWindow(windowStartMs, windowEndMs, dateTimeFieldSpec, timeColumn);
+ segmentProcessorConfigBuilder.setRecordFilterConfig(recordFilterConfig);
+
+ // Partitioner config from tableConfig
+ if (tableConfig.getIndexingConfig().getSegmentPartitionConfig() != null) {
+ Map<String, ColumnPartitionConfig> columnPartitionMap =
+ tableConfig.getIndexingConfig().getSegmentPartitionConfig().getColumnPartitionMap();
+ PartitionerConfig partitionerConfig = getPartitionerConfig(columnPartitionMap, tableNameWithType, schemaColumns);
+ segmentProcessorConfigBuilder.setPartitionerConfigs(Lists.newArrayList(partitionerConfig));
+ }
+
+ // Aggregations using configured Collector
+ List<String> sortedColumns = tableConfig.getIndexingConfig().getSortedColumn();
+ CollectorConfig collectorConfig =
+ getCollectorConfig(collectorTypeStr, aggregatorConfigs, schemaColumns, sortedColumns);
+ segmentProcessorConfigBuilder.setCollectorConfig(collectorConfig);
+
+ // Segment config
+ if (numRecordsPerSegment != null) {
+ SegmentConfig segmentConfig = getSegmentConfig(numRecordsPerSegment);
+ segmentProcessorConfigBuilder.setSegmentConfig(segmentConfig);
+ }
+
+ SegmentProcessorConfig segmentProcessorConfig = segmentProcessorConfigBuilder.build();
+
+ File inputSegmentsDir = new File(workingDir, INPUT_SEGMENTS_DIR);
+ Preconditions.checkState(inputSegmentsDir.mkdirs(), "Failed to create input directory: %s for task: %s",
+ inputSegmentsDir.getAbsolutePath(), taskType);
+ for (File indexDir : originalIndexDirs) {
+ FileUtils.copyDirectoryToDirectory(indexDir, inputSegmentsDir);
+ }
+ File outputSegmentsDir = new File(workingDir, OUTPUT_SEGMENTS_DIR);
+ Preconditions.checkState(outputSegmentsDir.mkdirs(), "Failed to create output directory: %s for task: %s",
+ outputSegmentsDir.getAbsolutePath(), taskType);
+
+ SegmentProcessorFramework segmentProcessorFramework =
+ new SegmentProcessorFramework(inputSegmentsDir, segmentProcessorConfig, outputSegmentsDir);
+ try {
+ segmentProcessorFramework.processSegments();
+ } finally {
+ segmentProcessorFramework.cleanup();
+ }
+
+ long endMillis = System.currentTimeMillis();
+ LOGGER.info("Finished task: {} with configs: {}. Total time: {}ms", taskType, configs, (endMillis - startMillis));
+ List<SegmentConversionResult> results = new ArrayList<>();
+ for (File file : outputSegmentsDir.listFiles()) {
+ String outputSegmentName = file.getName();
+ results.add(new SegmentConversionResult.Builder().setFile(file).setSegmentName(outputSegmentName)
+ .setTableNameWithType(tableNameWithType).build());
+ }
+ return results;
+ }
+
+ /**
+ * Construct a {@link RecordTransformerConfig} for time column transformation
+ */
+ private RecordTransformerConfig getRecordTransformerConfigForTime(String timeColumnTransformFunction,
+ String timeColumn) {
+ Map<String, String> transformationsMap = new HashMap<>();
+ transformationsMap.put(timeColumn, timeColumnTransformFunction);
+ return new RecordTransformerConfig.Builder().setTransformFunctionsMap(transformationsMap).build();
+ }
+
+ /**
+ * Construct a {@link RecordFilterConfig} by setting a filter function on the time column, for extracting data between window start/end
+ */
+ private RecordFilterConfig getRecordFilterConfigForWindow(long windowStartMs, long windowEndMs,
+ DateTimeFieldSpec dateTimeFieldSpec, String timeColumn) {
+ String filterFunction;
+ DateTimeFormatSpec dateTimeFormatSpec = new DateTimeFormatSpec(dateTimeFieldSpec.getFormat());
+ TimeUnit timeUnit = dateTimeFormatSpec.getColumnUnit();
+ DateTimeFieldSpec.TimeFormat timeFormat = dateTimeFormatSpec.getTimeFormat();
+ if (timeUnit.equals(TimeUnit.MILLISECONDS) && timeFormat.equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ // If time column is in EPOCH millis, use windowStart and windowEnd directly to filter
+ filterFunction = getFilterFunctionLong(windowStartMs, windowEndMs, timeColumn);
+ } else {
+ // Convert windowStart and windowEnd to time format of the data
+ if (dateTimeFormatSpec.getTimeFormat().equals(DateTimeFieldSpec.TimeFormat.EPOCH)) {
+ long windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, Long.class);
+ long windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, Long.class);
+ filterFunction = getFilterFunctionLong(windowStart, windowEnd, timeColumn);
+ } else {
+ String windowStart = dateTimeFormatSpec.fromMillisToFormat(windowStartMs, String.class);
+ String windowEnd = dateTimeFormatSpec.fromMillisToFormat(windowEndMs, String.class);
+ if (dateTimeFieldSpec.getDataType().isNumeric()) {
+ filterFunction = getFilterFunctionLong(Long.parseLong(windowStart), Long.parseLong(windowEnd), timeColumn);
+ } else {
+ filterFunction = getFilterFunctionString(windowStart, windowEnd, timeColumn);
+ }
+ }
+ }
+ return new RecordFilterConfig.Builder().setRecordFilterType(RecordFilterFactory.RecordFilterType.FILTER_FUNCTION)
+ .setFilterFunction(filterFunction).build();
+ }
+
+ /**
+ * Construct a {@link PartitionerConfig} using {@link org.apache.pinot.spi.config.table.SegmentPartitionConfig} from the table config
+ */
+ private PartitionerConfig getPartitionerConfig(Map<String, ColumnPartitionConfig> columnPartitionMap,
+ String tableNameWithType, Set<String> schemaColumns) {
+
Review comment:
remove line
----------------------------------------------------------------
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:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org