You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by GitBox <gi...@apache.org> on 2020/09/23 21:28:34 UTC
[GitHub] [druid] jihoonson commented on a change in pull request #10419: Automatically determine numShards for parallel ingestion hash partitioning
jihoonson commented on a change in pull request #10419:
URL: https://github.com/apache/druid/pull/10419#discussion_r493891466
##########
File path: docs/ingestion/native-batch.md
##########
@@ -294,11 +294,17 @@ How the worker task creates segments is:
|property|description|default|required?|
|--------|-----------|-------|---------|
|type|This should always be `hashed`|none|yes|
-|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|yes|
+|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|no|
|partitionDimensions|The dimensions to partition on. Leave blank to select all dimensions.|null|no|
The Parallel task with hash-based partitioning is similar to [MapReduce](https://en.wikipedia.org/wiki/MapReduce).
-The task runs in 2 phases, i.e., `partial segment generation` and `partial segment merge`.
+The task runs in up to 3 phases: `partial_dimension_cardinality`, `partial segment generation` and `partial segment merge`.
+- The `partial_dimension_cardinality` phase is an optional phase that only runs if `numShards` is not specified.
+The Parallel task splits the input data and assigns them to worker tasks based on the split hint spec.
+Each worker task (type `partial_dimension_cardinality`) gathers estimates of partitioning dimensions cardinality for
+each time chunk. The Parallel task will aggregate these estimates from the worker tasks and determine the highest
+cardinality across all of the time chunks in the input data, dividing this cardinality by `maxRowsPerSegment` to
Review comment:
It seems like that it will guarantee that you will not have segments than the computed `numShards`, but it will not be guaranteed that number of rows per segment doesn't exceed `maxRowsPerSegment` since the partition dimensions can be skewed. Is this correct? Then, I suggest `targetRowsPerSegment` since it's not a hard limit.
##########
File path: docs/ingestion/native-batch.md
##########
@@ -294,11 +294,17 @@ How the worker task creates segments is:
|property|description|default|required?|
|--------|-----------|-------|---------|
|type|This should always be `hashed`|none|yes|
-|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|yes|
+|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|no|
Review comment:
Please add the new parameter and its description.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialDimensionCardinalityTask.java
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.datasketches.hll.HllSketch;
+import org.apache.druid.data.input.HandlingInputRowIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexing.common.TaskToolbox;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
+import org.apache.druid.indexing.common.task.ClientBasedTaskInfoProvider;
+import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskResource;
+import org.apache.druid.indexing.common.task.batch.parallel.iterator.DefaultIndexTaskInputRowIteratorBuilder;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.granularity.GranularitySpec;
+import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PartialDimensionCardinalityTask extends PerfectRollupWorkerTask
+{
+ public static final String TYPE = "partial_dimension_cardinality";
+ private static final Logger LOG = new Logger(PartialDimensionCardinalityTask.class);
+
+ private final int numAttempts;
+ private final ParallelIndexIngestionSpec ingestionSchema;
+ private final String supervisorTaskId;
+
+ private final ObjectMapper jsonMapper;
+
+ @JsonCreator
+ PartialDimensionCardinalityTask(
+ // id shouldn't be null except when this task is created by ParallelIndexSupervisorTask
+ @JsonProperty("id") @Nullable String id,
+ @JsonProperty("groupId") final String groupId,
+ @JsonProperty("resource") final TaskResource taskResource,
+ @JsonProperty("supervisorTaskId") final String supervisorTaskId,
+ @JsonProperty("numAttempts") final int numAttempts, // zero-based counting
+ @JsonProperty("spec") final ParallelIndexIngestionSpec ingestionSchema,
+ @JsonProperty("context") final Map<String, Object> context,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(
+ getOrMakeId(id, TYPE, ingestionSchema.getDataSchema().getDataSource()),
+ groupId,
+ taskResource,
+ ingestionSchema.getDataSchema(),
+ ingestionSchema.getTuningConfig(),
+ context
+ );
+
+ Preconditions.checkArgument(
+ ingestionSchema.getTuningConfig().getPartitionsSpec() instanceof HashedPartitionsSpec,
+ "%s partitionsSpec required",
+ HashedPartitionsSpec.NAME
+ );
+
+ this.numAttempts = numAttempts;
+ this.ingestionSchema = ingestionSchema;
+ this.supervisorTaskId = supervisorTaskId;
+ this.jsonMapper = jsonMapper;
+ }
+
+ @JsonProperty
+ private int getNumAttempts()
+ {
+ return numAttempts;
+ }
+
+ @JsonProperty("spec")
+ private ParallelIndexIngestionSpec getIngestionSchema()
+ {
+ return ingestionSchema;
+ }
+
+ @JsonProperty
+ private String getSupervisorTaskId()
+ {
+ return supervisorTaskId;
+ }
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
+
+ @Override
+ public boolean isReady(TaskActionClient taskActionClient) throws Exception
+ {
+ return tryTimeChunkLock(
+ taskActionClient,
+ getIngestionSchema().getDataSchema().getGranularitySpec().inputIntervals()
+ );
+ }
+
+ @Override
+ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
+ {
+ DataSchema dataSchema = ingestionSchema.getDataSchema();
+ GranularitySpec granularitySpec = dataSchema.getGranularitySpec();
+ ParallelIndexTuningConfig tuningConfig = ingestionSchema.getTuningConfig();
+
+ HashedPartitionsSpec partitionsSpec = (HashedPartitionsSpec) tuningConfig.getPartitionsSpec();
+ Preconditions.checkNotNull(partitionsSpec, "partitionsSpec required in tuningConfig");
+
+ List<String> partitionDimensions = partitionsSpec.getPartitionDimensions();
+ if (partitionDimensions == null) {
+ partitionDimensions = HashBasedNumberedShardSpec.DEFAULT_PARTITION_DIMENSIONS;
+ }
+
+ InputSource inputSource = ingestionSchema.getIOConfig().getNonNullInputSource(
+ ingestionSchema.getDataSchema().getParser()
+ );
+ InputFormat inputFormat = inputSource.needsFormat()
+ ? ParallelIndexSupervisorTask.getInputFormat(ingestionSchema)
+ : null;
+ final RowIngestionMeters buildSegmentsMeters = toolbox.getRowIngestionMetersFactory().createRowIngestionMeters();
+ final ParseExceptionHandler parseExceptionHandler = new ParseExceptionHandler(
+ buildSegmentsMeters,
+ tuningConfig.isLogParseExceptions(),
+ tuningConfig.getMaxParseExceptions(),
+ tuningConfig.getMaxSavedParseExceptions()
+ );
+
+ try (
+ final CloseableIterator<InputRow> inputRowIterator = AbstractBatchIndexTask.inputSourceReader(
+ toolbox.getIndexingTmpDir(),
+ dataSchema,
+ inputSource,
+ inputFormat,
+ AbstractBatchIndexTask.defaultRowFilter(granularitySpec),
+ buildSegmentsMeters,
+ parseExceptionHandler
+ );
+ HandlingInputRowIterator iterator =
+ new DefaultIndexTaskInputRowIteratorBuilder()
Review comment:
nit: `DefaultIndexTaskInputRowIteratorBuilder` effectively does nothing here since its core functionality has been moved to `FilteringCloseableInputRowIterator` in #10336. I haven't cleaned up this interface yet. Now, it's only useful in range partitioning as some more useful inputRowHandlers are appended to the default builder.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialDimensionCardinalityTask.java
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.datasketches.hll.HllSketch;
+import org.apache.druid.data.input.HandlingInputRowIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexing.common.TaskToolbox;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
+import org.apache.druid.indexing.common.task.ClientBasedTaskInfoProvider;
+import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskResource;
+import org.apache.druid.indexing.common.task.batch.parallel.iterator.DefaultIndexTaskInputRowIteratorBuilder;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.granularity.GranularitySpec;
+import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PartialDimensionCardinalityTask extends PerfectRollupWorkerTask
+{
+ public static final String TYPE = "partial_dimension_cardinality";
+ private static final Logger LOG = new Logger(PartialDimensionCardinalityTask.class);
+
+ private final int numAttempts;
+ private final ParallelIndexIngestionSpec ingestionSchema;
+ private final String supervisorTaskId;
+
+ private final ObjectMapper jsonMapper;
+
+ @JsonCreator
+ PartialDimensionCardinalityTask(
+ // id shouldn't be null except when this task is created by ParallelIndexSupervisorTask
+ @JsonProperty("id") @Nullable String id,
+ @JsonProperty("groupId") final String groupId,
+ @JsonProperty("resource") final TaskResource taskResource,
+ @JsonProperty("supervisorTaskId") final String supervisorTaskId,
+ @JsonProperty("numAttempts") final int numAttempts, // zero-based counting
+ @JsonProperty("spec") final ParallelIndexIngestionSpec ingestionSchema,
+ @JsonProperty("context") final Map<String, Object> context,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(
+ getOrMakeId(id, TYPE, ingestionSchema.getDataSchema().getDataSource()),
+ groupId,
+ taskResource,
+ ingestionSchema.getDataSchema(),
+ ingestionSchema.getTuningConfig(),
+ context
+ );
+
+ Preconditions.checkArgument(
+ ingestionSchema.getTuningConfig().getPartitionsSpec() instanceof HashedPartitionsSpec,
+ "%s partitionsSpec required",
+ HashedPartitionsSpec.NAME
+ );
+
+ this.numAttempts = numAttempts;
+ this.ingestionSchema = ingestionSchema;
+ this.supervisorTaskId = supervisorTaskId;
+ this.jsonMapper = jsonMapper;
+ }
+
+ @JsonProperty
+ private int getNumAttempts()
+ {
+ return numAttempts;
+ }
+
+ @JsonProperty("spec")
+ private ParallelIndexIngestionSpec getIngestionSchema()
+ {
+ return ingestionSchema;
+ }
+
+ @JsonProperty
+ private String getSupervisorTaskId()
+ {
+ return supervisorTaskId;
+ }
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
+
+ @Override
+ public boolean isReady(TaskActionClient taskActionClient) throws Exception
+ {
+ return tryTimeChunkLock(
+ taskActionClient,
+ getIngestionSchema().getDataSchema().getGranularitySpec().inputIntervals()
+ );
+ }
+
+ @Override
+ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
+ {
+ DataSchema dataSchema = ingestionSchema.getDataSchema();
+ GranularitySpec granularitySpec = dataSchema.getGranularitySpec();
+ ParallelIndexTuningConfig tuningConfig = ingestionSchema.getTuningConfig();
+
+ HashedPartitionsSpec partitionsSpec = (HashedPartitionsSpec) tuningConfig.getPartitionsSpec();
+ Preconditions.checkNotNull(partitionsSpec, "partitionsSpec required in tuningConfig");
+
+ List<String> partitionDimensions = partitionsSpec.getPartitionDimensions();
+ if (partitionDimensions == null) {
+ partitionDimensions = HashBasedNumberedShardSpec.DEFAULT_PARTITION_DIMENSIONS;
+ }
+
+ InputSource inputSource = ingestionSchema.getIOConfig().getNonNullInputSource(
+ ingestionSchema.getDataSchema().getParser()
+ );
+ InputFormat inputFormat = inputSource.needsFormat()
+ ? ParallelIndexSupervisorTask.getInputFormat(ingestionSchema)
+ : null;
+ final RowIngestionMeters buildSegmentsMeters = toolbox.getRowIngestionMetersFactory().createRowIngestionMeters();
+ final ParseExceptionHandler parseExceptionHandler = new ParseExceptionHandler(
+ buildSegmentsMeters,
+ tuningConfig.isLogParseExceptions(),
+ tuningConfig.getMaxParseExceptions(),
+ tuningConfig.getMaxSavedParseExceptions()
+ );
+
+ try (
+ final CloseableIterator<InputRow> inputRowIterator = AbstractBatchIndexTask.inputSourceReader(
+ toolbox.getIndexingTmpDir(),
+ dataSchema,
+ inputSource,
+ inputFormat,
+ AbstractBatchIndexTask.defaultRowFilter(granularitySpec),
+ buildSegmentsMeters,
+ parseExceptionHandler
+ );
+ HandlingInputRowIterator iterator =
+ new DefaultIndexTaskInputRowIteratorBuilder()
+ .delegate(inputRowIterator)
+ .granularitySpec(granularitySpec)
+ .build()
+ ) {
+ Map<Interval, byte[]> cardinalities = determineCardinalities(
+ iterator,
+ granularitySpec,
+ partitionDimensions
+ );
+
+ sendReport(
+ toolbox,
+ new DimensionCardinalityReport(getId(), cardinalities)
+ );
+ }
+
+ return TaskStatus.success(getId());
+ }
+
+ private Map<Interval, byte[]> determineCardinalities(
+ HandlingInputRowIterator inputRowIterator,
+ GranularitySpec granularitySpec,
+ List<String> partitionDimensions
+ )
+ {
+ Map<Interval, HllSketch> intervalToCardinalities = new HashMap<>();
+ while (inputRowIterator.hasNext()) {
+ InputRow inputRow = inputRowIterator.next();
+ if (inputRow == null) {
+ continue;
+ }
+
+ DateTime timestamp = inputRow.getTimestamp();
+ //noinspection OptionalGetWithoutIsPresent (InputRowIterator returns rows with present intervals)
+ Interval interval = granularitySpec.bucketInterval(timestamp).get();
+
+ HllSketch hllSketch = intervalToCardinalities.computeIfAbsent(
+ interval,
+ (intervalKey) -> {
+ return DimensionCardinalityReport.createHllSketchForReport();
+ }
+ );
+ List<Object> groupKey = HashBasedNumberedShardSpec.getGroupKey(
+ partitionDimensions,
+ interval.getStartMillis(),
Review comment:
This timestamp should be bucketed based on the query granularity. See https://github.com/apache/druid/blob/master/core/src/main/java/org/apache/druid/timeline/partition/HashBasedNumberedShardSpec.java#L146.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
##########
@@ -582,6 +652,50 @@ private TaskStatus runRangePartitionMultiPhaseParallel(TaskToolbox toolbox) thro
return TaskStatus.fromCode(getId(), mergeState);
}
+ @VisibleForTesting
+ public static int determineNumShardsFromCardinalityReport(
+ Collection<DimensionCardinalityReport> reports,
+ int maxRowsPerSegment
+ )
+ {
+ // aggregate all the sub-reports
+ Map<Interval, Union> finalCollectors = new HashMap<>();
+ reports.forEach(report -> {
+ Map<Interval, byte[]> intervalToCardinality = report.getIntervalToCardinalities();
+ for (Map.Entry<Interval, byte[]> entry : intervalToCardinality.entrySet()) {
+ Union union = finalCollectors.computeIfAbsent(
+ entry.getKey(),
+ (key) -> {
+ return new Union(DimensionCardinalityReport.HLL_SKETCH_LOG_K);
+ }
+ );
+ HllSketch entryHll = HllSketch.wrap(Memory.wrap(entry.getValue()));
+ union.update(entryHll);
+ }
+ });
+
+ // determine the highest cardinality in any interval
+ long maxCardinality = Long.MIN_VALUE;
+ for (Union union : finalCollectors.values()) {
+ maxCardinality = Math.max(maxCardinality, (long) union.getEstimate());
+ }
+
+ LOG.info("Estimated max cardinality: " + maxCardinality);
+
+ // determine numShards based on maxRowsPerSegment and the highest per-interval cardinality
+ long numShards = maxCardinality / maxRowsPerSegment;
+ if (maxCardinality % maxRowsPerSegment != 0) {
+ // if there's a remainder add 1 so we stay under maxRowsPerSegment
+ numShards += 1;
+ }
+ try {
+ return Math.toIntExact(numShards);
+ }
+ catch (ArithmeticException ae) {
+ return Integer.MAX_VALUE;
Review comment:
Hmm.. Should we fail instead? Since the timeline in the coordinator and the broker will explode if you have this many segments per interval.
##########
File path: docs/ingestion/native-batch.md
##########
@@ -294,11 +294,17 @@ How the worker task creates segments is:
|property|description|default|required?|
|--------|-----------|-------|---------|
|type|This should always be `hashed`|none|yes|
-|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|yes|
+|numShards|Directly specify the number of shards to create. If this is specified and `intervals` is specified in the `granularitySpec`, the index task can skip the determine intervals/partitions pass through the data.|null|no|
|partitionDimensions|The dimensions to partition on. Leave blank to select all dimensions.|null|no|
The Parallel task with hash-based partitioning is similar to [MapReduce](https://en.wikipedia.org/wiki/MapReduce).
-The task runs in 2 phases, i.e., `partial segment generation` and `partial segment merge`.
+The task runs in up to 3 phases: `partial_dimension_cardinality`, `partial segment generation` and `partial segment merge`.
Review comment:
When I wrote this, my intention was showing the name of phases (e.g., https://github.com/apache/druid/pull/10419/files#diff-05bbc55d565a3d9462353d9b4771cb09R34). This phase name is not shown anywhere currently, but will be available in the task live reports and metrics after #10352. How about removing underscores from the here too?
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialDimensionCardinalityTask.java
##########
@@ -0,0 +1,258 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.druid.data.input.HandlingInputRowIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.hll.HyperLogLogCollector;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexing.common.TaskToolbox;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
+import org.apache.druid.indexing.common.task.ClientBasedTaskInfoProvider;
+import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskResource;
+import org.apache.druid.indexing.common.task.batch.parallel.iterator.DefaultIndexTaskInputRowIteratorBuilder;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.granularity.GranularitySpec;
+import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PartialDimensionCardinalityTask extends PerfectRollupWorkerTask
+{
+ public static final String TYPE = "partial_dimension_cardinality";
+ private static final Logger LOG = new Logger(PartialDimensionCardinalityTask.class);
+
+ private final int numAttempts;
+ private final ParallelIndexIngestionSpec ingestionSchema;
+ private final String supervisorTaskId;
+
+ private final ObjectMapper jsonMapper;
+
+ @JsonCreator
+ PartialDimensionCardinalityTask(
+ // id shouldn't be null except when this task is created by ParallelIndexSupervisorTask
+ @JsonProperty("id") @Nullable String id,
+ @JsonProperty("groupId") final String groupId,
+ @JsonProperty("resource") final TaskResource taskResource,
+ @JsonProperty("supervisorTaskId") final String supervisorTaskId,
+ @JsonProperty("numAttempts") final int numAttempts, // zero-based counting
+ @JsonProperty("spec") final ParallelIndexIngestionSpec ingestionSchema,
+ @JsonProperty("context") final Map<String, Object> context,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(
+ getOrMakeId(id, TYPE, ingestionSchema.getDataSchema().getDataSource()),
+ groupId,
+ taskResource,
+ ingestionSchema.getDataSchema(),
+ ingestionSchema.getTuningConfig(),
+ context
+ );
+
+ Preconditions.checkArgument(
+ ingestionSchema.getTuningConfig().getPartitionsSpec() instanceof HashedPartitionsSpec,
+ "%s partitionsSpec required",
+ HashedPartitionsSpec.NAME
+ );
+
+ this.numAttempts = numAttempts;
+ this.ingestionSchema = ingestionSchema;
+ this.supervisorTaskId = supervisorTaskId;
+ this.jsonMapper = jsonMapper;
+ }
+
+ @JsonProperty
+ private int getNumAttempts()
+ {
+ return numAttempts;
+ }
+
+ @JsonProperty("spec")
+ private ParallelIndexIngestionSpec getIngestionSchema()
+ {
+ return ingestionSchema;
+ }
+
+ @JsonProperty
+ private String getSupervisorTaskId()
+ {
+ return supervisorTaskId;
+ }
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
+
+ @Override
+ public boolean isReady(TaskActionClient taskActionClient) throws Exception
+ {
+ return tryTimeChunkLock(
+ taskActionClient,
+ getIngestionSchema().getDataSchema().getGranularitySpec().inputIntervals()
+ );
+ }
Review comment:
I think acquiring locks here should be fine since it's idempotent. The supervisor task and all its subtasks share the same lock based on their groupId. I actually think it's better to call `tryTimeChunkLock()` in every subtask since it will make the task fail early when its lock is revoked. Otherwise, the task will fail when it publishes segments which happens at the last stage in batch ingestion.
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/PartialDimensionCardinalityTask.java
##########
@@ -0,0 +1,254 @@
+/*
+ * 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.druid.indexing.common.task.batch.parallel;
+
+import com.fasterxml.jackson.annotation.JacksonInject;
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.google.common.base.Preconditions;
+import org.apache.datasketches.hll.HllSketch;
+import org.apache.druid.data.input.HandlingInputRowIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.indexer.TaskStatus;
+import org.apache.druid.indexer.partitions.HashedPartitionsSpec;
+import org.apache.druid.indexing.common.TaskToolbox;
+import org.apache.druid.indexing.common.actions.TaskActionClient;
+import org.apache.druid.indexing.common.task.AbstractBatchIndexTask;
+import org.apache.druid.indexing.common.task.ClientBasedTaskInfoProvider;
+import org.apache.druid.indexing.common.task.IndexTask;
+import org.apache.druid.indexing.common.task.TaskResource;
+import org.apache.druid.indexing.common.task.batch.parallel.iterator.DefaultIndexTaskInputRowIteratorBuilder;
+import org.apache.druid.java.util.common.logger.Logger;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.apache.druid.segment.incremental.ParseExceptionHandler;
+import org.apache.druid.segment.incremental.RowIngestionMeters;
+import org.apache.druid.segment.indexing.DataSchema;
+import org.apache.druid.segment.indexing.granularity.GranularitySpec;
+import org.apache.druid.timeline.partition.HashBasedNumberedShardSpec;
+import org.joda.time.DateTime;
+import org.joda.time.Interval;
+
+import javax.annotation.Nullable;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+public class PartialDimensionCardinalityTask extends PerfectRollupWorkerTask
+{
+ public static final String TYPE = "partial_dimension_cardinality";
+ private static final Logger LOG = new Logger(PartialDimensionCardinalityTask.class);
+
+ private final int numAttempts;
+ private final ParallelIndexIngestionSpec ingestionSchema;
+ private final String supervisorTaskId;
+
+ private final ObjectMapper jsonMapper;
+
+ @JsonCreator
+ PartialDimensionCardinalityTask(
+ // id shouldn't be null except when this task is created by ParallelIndexSupervisorTask
+ @JsonProperty("id") @Nullable String id,
+ @JsonProperty("groupId") final String groupId,
+ @JsonProperty("resource") final TaskResource taskResource,
+ @JsonProperty("supervisorTaskId") final String supervisorTaskId,
+ @JsonProperty("numAttempts") final int numAttempts, // zero-based counting
+ @JsonProperty("spec") final ParallelIndexIngestionSpec ingestionSchema,
+ @JsonProperty("context") final Map<String, Object> context,
+ @JacksonInject ObjectMapper jsonMapper
+ )
+ {
+ super(
+ getOrMakeId(id, TYPE, ingestionSchema.getDataSchema().getDataSource()),
+ groupId,
+ taskResource,
+ ingestionSchema.getDataSchema(),
+ ingestionSchema.getTuningConfig(),
+ context
+ );
+
+ Preconditions.checkArgument(
+ ingestionSchema.getTuningConfig().getPartitionsSpec() instanceof HashedPartitionsSpec,
+ "%s partitionsSpec required",
+ HashedPartitionsSpec.NAME
+ );
+
+ this.numAttempts = numAttempts;
+ this.ingestionSchema = ingestionSchema;
+ this.supervisorTaskId = supervisorTaskId;
+ this.jsonMapper = jsonMapper;
+ }
+
+ @JsonProperty
+ private int getNumAttempts()
+ {
+ return numAttempts;
+ }
+
+ @JsonProperty("spec")
+ private ParallelIndexIngestionSpec getIngestionSchema()
+ {
+ return ingestionSchema;
+ }
+
+ @JsonProperty
+ private String getSupervisorTaskId()
+ {
+ return supervisorTaskId;
+ }
+
+ @Override
+ public String getType()
+ {
+ return TYPE;
+ }
+
+ @Override
+ public boolean isReady(TaskActionClient taskActionClient) throws Exception
+ {
+ return tryTimeChunkLock(
+ taskActionClient,
+ getIngestionSchema().getDataSchema().getGranularitySpec().inputIntervals()
+ );
+ }
+
+ @Override
+ public TaskStatus runTask(TaskToolbox toolbox) throws Exception
+ {
+ DataSchema dataSchema = ingestionSchema.getDataSchema();
+ GranularitySpec granularitySpec = dataSchema.getGranularitySpec();
+ ParallelIndexTuningConfig tuningConfig = ingestionSchema.getTuningConfig();
+
+ HashedPartitionsSpec partitionsSpec = (HashedPartitionsSpec) tuningConfig.getPartitionsSpec();
+ Preconditions.checkNotNull(partitionsSpec, "partitionsSpec required in tuningConfig");
+
+ List<String> partitionDimensions = partitionsSpec.getPartitionDimensions();
+ if (partitionDimensions == null) {
+ partitionDimensions = HashBasedNumberedShardSpec.DEFAULT_PARTITION_DIMENSIONS;
+ }
+
+ InputSource inputSource = ingestionSchema.getIOConfig().getNonNullInputSource(
+ ingestionSchema.getDataSchema().getParser()
+ );
+ InputFormat inputFormat = inputSource.needsFormat()
+ ? ParallelIndexSupervisorTask.getInputFormat(ingestionSchema)
+ : null;
+ final RowIngestionMeters buildSegmentsMeters = toolbox.getRowIngestionMetersFactory().createRowIngestionMeters();
+ final ParseExceptionHandler parseExceptionHandler = new ParseExceptionHandler(
+ buildSegmentsMeters,
+ tuningConfig.isLogParseExceptions(),
+ tuningConfig.getMaxParseExceptions(),
+ tuningConfig.getMaxSavedParseExceptions()
+ );
+
+ try (
+ final CloseableIterator<InputRow> inputRowIterator = AbstractBatchIndexTask.inputSourceReader(
+ toolbox.getIndexingTmpDir(),
+ dataSchema,
+ inputSource,
+ inputFormat,
+ AbstractBatchIndexTask.defaultRowFilter(granularitySpec),
+ buildSegmentsMeters,
+ parseExceptionHandler
+ );
+ HandlingInputRowIterator iterator =
+ new DefaultIndexTaskInputRowIteratorBuilder()
+ .delegate(inputRowIterator)
+ .granularitySpec(granularitySpec)
+ .build()
+ ) {
+ Map<Interval, byte[]> cardinalities = determineCardinalities(
+ iterator,
+ granularitySpec,
+ partitionDimensions
+ );
+
+ sendReport(
+ toolbox,
+ new DimensionCardinalityReport(getId(), cardinalities)
+ );
+ }
+
+ return TaskStatus.success(getId());
+ }
+
+ private Map<Interval, byte[]> determineCardinalities(
+ HandlingInputRowIterator inputRowIterator,
+ GranularitySpec granularitySpec,
+ List<String> partitionDimensions
+ )
+ {
+ Map<Interval, HllSketch> intervalToCardinalities = new HashMap<>();
+ while (inputRowIterator.hasNext()) {
+ InputRow inputRow = inputRowIterator.next();
+ if (inputRow == null) {
Review comment:
nit: `inputRow` is safely non-null since `FilteringCloseableInputRowIterator` filters out all null rows. See `AbstractBatchIndexTask.defaultRowFilter()`.
----------------------------------------------------------------
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@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org