[
https://issues.apache.org/jira/browse/TEZ-3207?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15246770#comment-15246770
]
Siddharth Seth commented on TEZ-3207:
-------------------------------------
One thing that is likely not covered is that ShuffleManager does not know how
many partitions are coming in for an Input. If the events arrive with a delay
(e.g. crossing the 500 limit of #events per heartbeat, or the fetcher is faster
than processing of events) - we may end up missing data.
How is the AM side plugin setting up #phyiscalInputs ?
Other comments, which are mostly independent of the core functionality change.
- Rename hasNumPendingPartitions to getNumPendingPartitions since it is not
returning a boolean
- addKnownInput - can be refactored with inputs =
partitionToInputs.get(partition); being further up to avoid one unnecessary map
lookup
- clearAndGetOnePartition - Iterator + null check instead of a for loop. Ignore
if you want - doesn't make a difference.
- toString in InputHost - can we include the numPending partitions, and num
pending segments within that
- visibleForTesting annotation on constructFetcherForHost
- host = new InputHost(identifier); could be host = identifier ?
- On the test - can we reduce the sleep itnerval since this is a poll. It will
normally complete well before the 2s sleep time. Ideally, we should change the
test to control the thread on the main ShuffleManager - but that can be done
later.
- Would be useful to add a test where all the events don't arrive at the same
time. i.e. events for partition1 arrive first. Ensure Shuffle does not
complete, and keeps waiting for the remaining partitions.
Will look in a little more detail later.
> Add support for fetching multiple partitions from the same source task to
> UnorderedKVInput
> ------------------------------------------------------------------------------------------
>
> Key: TEZ-3207
> URL: https://issues.apache.org/jira/browse/TEZ-3207
> Project: Apache Tez
> Issue Type: Bug
> Reporter: Ming Ma
> Assignee: Ming Ma
> Attachments: TEZ-3207.patch
>
>
> The ordered grouped {{ShuffleScheduler}} can support fetching multiple
> partitions from the same source task. But for the unordered ShuffleManager,
> it only supports one partition per source task due to the following issue
> where {{identifier}} doesn't take partition id into account.
> {noformat}
> public void addKnownInput(String hostName, int port,
> InputAttemptIdentifier srcAttemptIdentifier, int srcPhysicalIndex) {
> String identifier = InputHost.createIdentifier(hostName, port);
> InputHost host = knownSrcHosts.get(identifier);
> ....
> }
> {noformat}
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
