Sub-partitioning supports repartitioning the input data multiple times #7996
Conversation
Signed-off-by: Firestarman <[email protected]>
Signed-off-by: Firestarman <[email protected]>
Signed-off-by: Firestarman <[email protected]>
|
build |
| * Always return the current seed. | ||
| * This is intended to share the same seed across sub-partitioners. | ||
| */ | ||
| override final def nextSeed: Int = seedGenerator.currentSeed |
There was a problem hiding this comment.
nit: This is confusing. I would prefer it if we had a way to pass the seed directly into a sub-partitioner instead.
| } | ||
|
|
||
| private[this] def needRepartition(parts: SubPartitionBuffer): Boolean = { | ||
| // FIXME Is it good enough to ask for repartitioning when there exists any sub |
There was a problem hiding this comment.
I think we would want to stop for a given batch if we try to partition the batch and it didn't change. Meaning we tried to partition it N ways and we got back N-1 empty batches and 1 batch with all of the data in it.
That said I think we only need to do two passes through the data at most. The algorithm I see would be something like the following.
- If the build side < target batch size
- do the join as normal
- else
- partition the build/stream sides according to the configured number of splits
- group build partitions to try and be just under target batch size and do the join for all build partitions that met this goal
- for each build partition > target batch size
- num_partitions = ceil(build size / target batch size)
- partition the build and stream batches using the new seed and the calculated number of partitions
- group partitions together again like before and do the join no matter how large the size is.
sameerz
added
the
reliability
Signed-off-by: Firestarman <[email protected]>
Signed-off-by: Firestarman <[email protected]>
|
build |
| pair = Some(new PartitionPair(buildBatch, streamBatches)) | ||
| } | ||
| } | ||
| } else if (bigBuildBatches.nonEmpty) { |
There was a problem hiding this comment.
the way this seems to be working is that big batches are handled later, after the original sub partitioner iterators are flushed. Do I understand that correctly? If so I wonder if it makes sense to stop pulling from the sub partitioner, and give priority to the big batches temporarily, because that should alleviate the need to spill all the batches in bigBuildBatches.
There was a problem hiding this comment.
Yes the big batches come later. That is what I suggested we do above #7996 (comment)
They come later because it allows us to know their size and repartition them in a single pass instead of needing to possibly split it more than twice. I think we might be able to do something similar to what you want. But we would still have to process the entire build side first. Once we have pulled in the build side and repartitioned it the first time, we could then know which batches we would want to repartition a second time and how many sub-partitions we would need. But then we have to hand the same plan over to the steam side partitioner. We would have to be able to tell it what the seed is for the first partition pass and how many partitions to use, along with what partitions would need a second pass and the corresponding seeds and number of partitions for each of them.
It would be doable, but it is a fairly large change, and it would be a performance improvement when we are in a memory constrained situation. So we would want to have a few queries that exhibited this behavior that we could use to benchmark it.
That all sounds rather involved and I think it would be best to just file a follow on issue to look at it and we can the prioritize it in the backlog according to what management decides.
There was a problem hiding this comment.
This sounds good to me. I filed #8057
NVIDIA#7996) Signed-off-by: Firestarman <[email protected]>
closes #7911
This adds in the support of repartitioning the input data multiple times with different hash seeds to sub-partitioning when the initial partition number is not big enough to over partition data. It will also calculate the actual partition number needed for the over partitioning.