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/11/18 02:26:31 UTC
[GitHub] [druid] jihoonson opened a new pull request #10592: Allow missing intervals for Parallel task
jihoonson opened a new pull request #10592:
URL: https://github.com/apache/druid/pull/10592
### Description
This PR allows Parallel task to run without explicit intervals in granularitySpec. If intervals are missing, the parallel task executes an extra step for input sampling which collects the intervals to index.
This PR additionally fixes a bug when `numShards` is missing in hash partitioning. When `numShards` is missing, the parallel task computes it by scanning the whole input. However, the computed numShards was ignored when it's serialized into JSON. To fix it, this PR adds another field `intervalToNumShardsOverride` which stores the computed numShards per interval so that we can handle data skew well across intervals.
<hr>
This PR has:
- [x] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [x] added documentation for new or modified features or behaviors.
- [x] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [ ] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/licenses.yaml)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson merged pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
jihoonson merged pull request #10592:
URL: https://github.com/apache/druid/pull/10592
----------------------------------------------------------------
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
[GitHub] [druid] abhishekagarwal87 commented on a change in pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
abhishekagarwal87 commented on a change in pull request #10592:
URL: https://github.com/apache/druid/pull/10592#discussion_r525876638
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
##########
@@ -541,9 +561,12 @@ private TaskStatus runHashPartitionMultiPhaseParallel(TaskToolbox toolbox) throw
);
}
- final Integer numShardsOverride;
+ final Map<Interval, Integer> intervalToNumShards;
HashedPartitionsSpec partitionsSpec = (HashedPartitionsSpec) ingestionSchema.getTuningConfig().getPartitionsSpec();
- if (partitionsSpec.getNumShards() == null) {
+ final boolean needsInputSampling =
+ partitionsSpec.getNumShards() == null
+ || ingestionSchemaToUse.getDataSchema().getGranularitySpec().inputIntervals().isEmpty();
+ if (needsInputSampling) {
// 0. need to determine numShards by scanning the data
LOG.info("numShards is unspecified, beginning %s phase.", PartialDimensionCardinalityTask.TYPE);
Review comment:
this log statement needs a change now
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
##########
@@ -552,38 +575,50 @@ private TaskStatus runHashPartitionMultiPhaseParallel(TaskToolbox toolbox) throw
this::createPartialDimensionCardinalityRunner
);
- if (cardinalityRunner == null) {
- throw new ISE("Could not create cardinality runner for hash partitioning.");
- }
-
state = runNextPhase(cardinalityRunner);
if (state.isFailure()) {
return TaskStatus.failure(getId());
}
- int effectiveMaxRowsPerSegment = partitionsSpec.getMaxRowsPerSegment() == null
- ? PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT
- : partitionsSpec.getMaxRowsPerSegment();
- LOG.info("effective maxRowsPerSegment is: " + effectiveMaxRowsPerSegment);
+ if (cardinalityRunner.getReports().isEmpty()) {
+ String msg = "No valid rows for hash partitioning."
+ + " All rows may have invalid timestamps or have been filtered out.";
+ LOG.warn(msg);
+ return TaskStatus.success(getId(), msg);
+ }
+
+ if (partitionsSpec.getNumShards() == null) {
+ int effectiveMaxRowsPerSegment = partitionsSpec.getMaxRowsPerSegment() == null
+ ? PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT
+ : partitionsSpec.getMaxRowsPerSegment();
+ LOG.info("effective maxRowsPerSegment is: " + effectiveMaxRowsPerSegment);
- if (cardinalityRunner.getReports() == null) {
- throw new ISE("Could not determine cardinalities for hash partitioning.");
+ intervalToNumShards = determineNumShardsFromCardinalityReport(
+ cardinalityRunner.getReports().values(),
+ effectiveMaxRowsPerSegment
+ );
+ } else {
+ intervalToNumShards = CollectionUtils.mapValues(
+ mergeCardinalityReports(cardinalityRunner.getReports().values()),
+ k -> partitionsSpec.getNumShards()
+ );
}
- numShardsOverride = determineNumShardsFromCardinalityReport(
- cardinalityRunner.getReports().values(),
- effectiveMaxRowsPerSegment
- );
- LOG.info("Automatically determined numShards: " + numShardsOverride);
Review comment:
is there an equivalent info being logged somewhere now?
----------------------------------------------------------------
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
[GitHub] [druid] abhishekagarwal87 commented on pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
abhishekagarwal87 commented on pull request #10592:
URL: https://github.com/apache/druid/pull/10592#issuecomment-733478653
> @abhishekagarwal87 do you have more comments?
@jihoonson LGTM
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
jihoonson commented on pull request #10592:
URL: https://github.com/apache/druid/pull/10592#issuecomment-733979702
@clintropolis @himanshug @abhishekagarwal87 thanks for the review :+1:
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on a change in pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
jihoonson commented on a change in pull request #10592:
URL: https://github.com/apache/druid/pull/10592#discussion_r526366690
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
##########
@@ -541,9 +561,12 @@ private TaskStatus runHashPartitionMultiPhaseParallel(TaskToolbox toolbox) throw
);
}
- final Integer numShardsOverride;
+ final Map<Interval, Integer> intervalToNumShards;
HashedPartitionsSpec partitionsSpec = (HashedPartitionsSpec) ingestionSchema.getTuningConfig().getPartitionsSpec();
- if (partitionsSpec.getNumShards() == null) {
+ final boolean needsInputSampling =
+ partitionsSpec.getNumShards() == null
+ || ingestionSchemaToUse.getDataSchema().getGranularitySpec().inputIntervals().isEmpty();
+ if (needsInputSampling) {
// 0. need to determine numShards by scanning the data
LOG.info("numShards is unspecified, beginning %s phase.", PartialDimensionCardinalityTask.TYPE);
Review comment:
:+1:
##########
File path: indexing-service/src/main/java/org/apache/druid/indexing/common/task/batch/parallel/ParallelIndexSupervisorTask.java
##########
@@ -552,38 +575,50 @@ private TaskStatus runHashPartitionMultiPhaseParallel(TaskToolbox toolbox) throw
this::createPartialDimensionCardinalityRunner
);
- if (cardinalityRunner == null) {
- throw new ISE("Could not create cardinality runner for hash partitioning.");
- }
-
state = runNextPhase(cardinalityRunner);
if (state.isFailure()) {
return TaskStatus.failure(getId());
}
- int effectiveMaxRowsPerSegment = partitionsSpec.getMaxRowsPerSegment() == null
- ? PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT
- : partitionsSpec.getMaxRowsPerSegment();
- LOG.info("effective maxRowsPerSegment is: " + effectiveMaxRowsPerSegment);
+ if (cardinalityRunner.getReports().isEmpty()) {
+ String msg = "No valid rows for hash partitioning."
+ + " All rows may have invalid timestamps or have been filtered out.";
+ LOG.warn(msg);
+ return TaskStatus.success(getId(), msg);
+ }
+
+ if (partitionsSpec.getNumShards() == null) {
+ int effectiveMaxRowsPerSegment = partitionsSpec.getMaxRowsPerSegment() == null
+ ? PartitionsSpec.DEFAULT_MAX_ROWS_PER_SEGMENT
+ : partitionsSpec.getMaxRowsPerSegment();
+ LOG.info("effective maxRowsPerSegment is: " + effectiveMaxRowsPerSegment);
- if (cardinalityRunner.getReports() == null) {
- throw new ISE("Could not determine cardinalities for hash partitioning.");
+ intervalToNumShards = determineNumShardsFromCardinalityReport(
+ cardinalityRunner.getReports().values(),
+ effectiveMaxRowsPerSegment
+ );
+ } else {
+ intervalToNumShards = CollectionUtils.mapValues(
+ mergeCardinalityReports(cardinalityRunner.getReports().values()),
+ k -> partitionsSpec.getNumShards()
+ );
}
- numShardsOverride = determineNumShardsFromCardinalityReport(
- cardinalityRunner.getReports().values(),
- effectiveMaxRowsPerSegment
- );
- LOG.info("Automatically determined numShards: " + numShardsOverride);
Review comment:
No, I didn't add one for `intervalToNumShards` because it could have lots of intervals.
----------------------------------------------------------------
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
[GitHub] [druid] vogievetsky commented on pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
vogievetsky commented on pull request #10592:
URL: https://github.com/apache/druid/pull/10592#issuecomment-729345908
No comment on the code itself, but very exciting to remove the extra requirement from the user!
----------------------------------------------------------------
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
[GitHub] [druid] jihoonson commented on pull request #10592: Allow missing intervals for Parallel task with hash/range partitioning
Posted by GitBox <gi...@apache.org>.
jihoonson commented on pull request #10592:
URL: https://github.com/apache/druid/pull/10592#issuecomment-730688213
@abhishekagarwal87 do you have more comments?
----------------------------------------------------------------
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