[jira] [Comment Edited] (DRILL-4706) Fragment planning causes Drillbits to read remote chunks when local copies are available

["Padma Penumarthy (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/DRILL-4706?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15623642#comment-15623642 ] 

Padma Penumarthy edited comment on DRILL-4706 at 12/3/16 12:44 AM:
-------------------------------------------------------------------

Notes about how current parallelization algorithms(SoftAffinity and HardAffinity based) work, why remote reads are happening  and details of new algorithm (LocalAffinity based) implemented.

Most operators do not have any affinity i.e. they can be scheduled on any node. GroupScan and Store are the operators which have affinity i.e. they implement HasAffinity interface. Operators which have affinity convey their affinity preference through getDistributionAffinity() interface. Affinity for the fragment,  which decides how the fragment is going to be parallelized is decided by the operator with most restricted affinity in the fragment. Restriction is enforced in the following order: NONE, SOFT and HARD. We have two parallelization algorithms (SOFT and HARD). NONE and SOFT use SoftAffinityFragmentParallelizer and HARD uses HardAffinityFragmentParallelizer. 

Screen (which implements Store) and distributed system tables(memory, threads) use hard affinity. 
Group Scan and system tables (boot, drill bits and version) use soft affinity.

If we have screen (hard affinity) and scan (soft affinity) in the same fragment, affinity for the fragment will be hard since that is the more restrictive of the two. 

SoftAffinity (current algorithm for scheduling parquet scan fragments):
Initialization (getPlan -> GetGroupScan -> ParquetGroupScan.init):
1.When parquet metadata is read, for each rowGroup,
HostAffinity for each host (ratio of number of bytes present on that host / total bytes for the rowGroup) is calculated.
2. EndPointAffinity for each host (ratio of number of bytes on the host/total bytes  for the whole scan) is calculated.

Parallelize the scan (SoftAffinityFragmentParallelizer.parallelizeFragment):
1. Compute how many total fragments to schedule (width)  (Based on cost, slice target, min and maxWidth for the operator, maxGlobalWidth, maxWidthPerNode etc.)
2. Divide by number of nodes  - This is the average number of fragments we want to run on each node.
3. To favor nodes with affinity > 0 i.e nodes that have some local data (does not matter what the value is), 
    multiply the value from 2 above by affinity factor  - This is the number of fragments we want to schedule on each node with affinity.
4. Schedule upto  number of fragments calculated from 3 above on each of the nodes with affinity in round robin fashion.
5. If we schedule the required number of fragments (i.e. width from 1 above), we are done.
6. Else, rest of fragments, we schedule on nodes which do not have any local data i.e. nodes with no affinity in a round robin fashion.

Assignment(AssignmentCreator.getMappings):
1. To distribute rowGroups uniformly, calculate maxWork each fragment should do (total number of rowGroups/total number of fragments)
2.  For each endpoint, calculate the maxCount (maxWork * number of fragments on the endpoint) and minCount (at least 1 per fragment or (maxWork-1) * number of fragments) number of rowGroups to assign. 
3. Assign up to minCount rowGroups per endpoint in a round robin fashion, selecting from the sorted list of hosts(sorted based on host affinity) for each rowGroup.
4. If there are any leftovers, assign to the endPoints which do not have minimum (i.e. minCount) assigned yet.
5. If there are still leftovers, assign to the endPoints which do not maximum (i.e. maxCount) assigned yet.

Why is this causing remote reads and why increasing affinity factor does not help ?
When the data is skewed i.e. data is not distributed equally, all nodes with affinity still get equal number of fragments assigned (because they have some data). 
We are not assigning fragments proportional to affinity value i.e. amount of data available on the node.
So, some of them have to do remote read since data is not available locally. Since they all are treated equally,
increasing affinity factor does not help. affinity factor only helps in eliminating nodes which do not
have any data vs. nodes which have some data.

Another problem  is calculation of endpoint affinity values. We do not take replication factor into account and end up including bytes for a rowGroup multiple times on different hosts. Based on data distribution, this results in skewed affinity values which do not reflect how those values are being/should be used. 

HardAffinity Algorithm:
HardAffinity assigns fragments only to nodes which are marked as mandatory in the endpointAffinity. It works as follows:
1. Add nodes with endpointAffinities marked as assignmentRequired (i.e. mandatory) to the endpointPool.
2 .Calculate how many fragments to schedule (width) (Based on cost, slice target, min and maxWidth for the fragment, maxGlobalWidth, maxWidthPerNode and maxWidth specified in endpointAffinities)
3. Note that  width is constrained by maxWidthPerNode * endpointPool.size() (1 above) and sum of maxWidths specified by each endpointAffinity entry.
4.  Schedule at least 1 fragment on each endpoint in the endpointPool (since they all are marked as mandatory)
5.  For the remaining slots i.e. (width - endpointPool.size()), assign fragments to endpoints proportional to their affinity constrained by maxWidthPerNode and endpointAffinity maxWidth.

LocalAffinity (new algorithm based on locality of data):
This is not enabled by default. To use the new algorithm, we need to set system option `parquet.use_local_affinity`=true. Every effort is made to have the new code under the new option so no regressions are introduced. 
This will invoke a new local affinity fragment parallelizer which is less restrictive than soft affinity fragment parallelizer and is enabled only for parquet group scan.

Initialization(getPlan -> GetGroupScan -> ParquetGroupScan.init):
1. When parquet metadata is read, for each rowGroup, we need to compute the best possible host to scan it on (computeRowGroupAssignment)
2. For each rowGroup, get the lists of hosts which have maximum data available locally for the rowGroup (topEndpoints).
3. From that list, pick the node which has minimum amount of work assigned so far (based on number of bytes assigned to scan on that node).
4. Repeat 2 and 3  for second pass so we make adjustments after one round of allocations are done i.e. after first iteration.
5. Once we compute the best possible node on which to scan the rowGroup, save that information (preferredEndpoint). Note: preferredEndpoint will be null if there is no drillbit running on any of the nodes which have data or if it is local file system. 
6. Update endpointAffinity for each node with the number of rowGroups (numLocalWorkUnits) assigned to be scanned on that endpoint.

Parallelize the Scan(LocalAffinityFragmentParallelizer.parallelizeFragment):
1. Decide how many total fragments to run (width) (Based on cost, slice target, min and maxWidth for the operator, maxGlobalWidth, maxWidthPerNode etc.)
2. Include each endpoint which has affinity with numLocalWorkUnits > 0 in the list of endpoints on which we want to schedule the fragments (localEndpointPool).  
3. Since we want to assign only to nodes with data locality, constrain the number calculated in 1 above to maximum that can be assigned to nodes with locality to be localWidth (maxWidthPerNode * localEndpointPool.size())
4. Sort the endpoints in localEndpointPool based on work they have to do i.e. numLocalWorkUnits
5. Calculate how many fragments to assign to each of the nodes in localEndpointPool based on how much work they have to do i.e. targetAllocation (proportional to numLocalWorkUnits assigned to the node).
6. Assign one fragment to each of the nodes from the localEndpointPool to make sure minimum of one is assigned to each of them.
7. Go through the sorted localEndpointPool in a round robin way and keep assigning fragments to individual nodes till their target allocation or maxWidthPerNode is reached.
8. For 6 and 7, Stop when overall allocation reaches the total target (localWidth).
9. At this point, we have taken care of allocating fragments for totalLocalWorkUnits, i.e. workUnits which have data locality information.
10. If we have assigned fragments for all workUnits, we are done.  
11. It is possible that some workUnits have preferred endPoints null (because there is no drill bit running on the hosts which have data for the workUnit). In that case, we will have unassigned work Items.
12. Allocate the fragments for unassigned work items to active end points, making sure maxWidthPerNode constraint is honored.

Assignment(AssignmentCreator.getMappings):
1. When the system option parquet.use_local_affinity is set to true, assign each rowGroup to a fragment (round robin) on it’s preferredEndPoint (assignLocal).
2. If the preferredEndpoint is null or fragment is not available on that node, add it to unassignedList
3. If we have unassigned work items, first assign at least one work item to fragments which have nothing assigned so we meet the minimum requirement.
4. Assign any remaining unassigned work items to fragments in a round robin way.



was (Author: ppenumarthy):
Notes about how current algorithm(SoftAffinity based) works, why remote reads happen and the new algorithm (LocalAffinity based) implemented.

SoftAffinity (current algorithm for scheduling parquet scan fragments):

Initialization (getPlan -> GetGroupScan -> ParquetGroupScan.init):
1.When parquet metadata is read, for each rowGroup,
HostAffinity for each host (ratio of number of bytes present on that host / total bytes for the rowGroup) is calculated.
2. EndPointAffinity for each host (ratio of number of bytes on the host/total bytes  for the whole scan) is calculated.

Parallelize the scan (SoftAffinityFragmentParallelizer.parallelizeFragment):
1. Compute how many total fragments to schedule (width)  (Based on cost, slice target, min and maxWidth for the operator, maxGlobalWidth, maxWidthPerNode etc.)
2. Divide by number of nodes  - This is the average number of fragments we want to run on each node.
3. To favor nodes with affinity > 0 i.e nodes that have some local data (does not matter what the value is), 
    multiply the value from 2 above by affinity factor  - This is the number of fragments we want to schedule on each node with affinity.
4. Schedule upto  number of fragments calculated from 3 above on each of the nodes with affinity in round robin fashion.
5. If we schedule the required number of fragments (i.e. width from 1 above), we are done.
6. Else, rest of fragments, we schedule on nodes which do not have any local data i.e. nodes with no affinity in a round robin fashion.

Assignment(AssignmentCreator.getMappings):
1. To distribute rowGroups uniformly, calculate maxWork each fragment should do (total number of rowGroups/total number of fragments)
2.  For each endpoint, calculate the maxCount (maxWork * number of fragments on the endpoint) and minCount (at least 1 per fragment or (maxWork-1) * number of fragments) number of rowGroups to assign. 
3. Assign up to minCount rowGroups per endpoint in a round robin fashion, selecting from the sorted list of hosts(sorted based on host affinity) for each rowGroup.
4. If there are any leftovers, assign to the endPoints which do not have minimum (i.e. minCount) assigned yet.
5. If there are still leftovers, assign to the endPoints which do not maximum (i.e. maxCount) assigned yet.

Why is this causing remote reads and why increasing affinity factor does not help ?
When the data is skewed i.e. data is not distributed equally, all nodes with affinity still get equal number of fragments assigned (because they have some data). 
We are not assigning fragments proportional to affinity value i.e. amount of data available on the node.
So, some of them have to do remote read since data is not available locally. Since they all are treated equally,
increasing affinity factor does not help. affinity factor only helps in eliminating nodes which do not
have any data vs. nodes which have some data.

Another problem  is calculation of endpoint affinity values. We do not take replication factor into account and end up including bytes for a rowGroup multiple times on different hosts. Based on data distribution, this results in skewed affinity values which do not reflect how those values are being/should be used. 


LocalAffinity (new algorithm based on locality of data):
This is not enabled by default. To use the new algorithm, we need to set system option `parquet.use_local_affinity`=true. Every effort is made to have the new code under the new option so no regressions are introduced. 
This will invoke a new local affinity fragment parallelizer which is less restrictive than soft affinity fragment parallelizer and is enabled only for parquet group scan.

Initialization(getPlan -> GetGroupScan -> ParquetGroupScan.init):
1. When parquet metadata is read, for each rowGroup, we need to compute the best possible host to scan it on (computeRowGroupAssignment)
2. For each rowGroup, get the lists of hosts which have maximum data available locally for the rowGroup (topEndpoints).
3. From that list, pick the node which has minimum amount of work assigned so far (based on number of bytes assigned to scan on that node).
4. Repeat 2 and 3  for second pass so we make adjustments after one round of allocations are done i.e. after first iteration.
5. Once we compute the best possible node on which to scan the rowGroup, save that information (preferredEndpoint). Note: preferredEndpoint will be null if there is no drillbit running on any of the nodes which have data or if it is local file system. 
6. Update endpointAffinity for each node with the number of rowGroups (localWorkUnits) assigned to be scanned on that endpoint.

Parallelize the Scan(LocalAffinityFragmentParallelizer.parallelizeFragment):
1. Decide how many total fragments to run (width) (Based on cost, slice target, min and maxWidth for the operator, maxGlobalWidth, maxWidthPerNode etc.)
2. Include each endpoint which has affinity with localWorkUnits > 0 in the list of endpoints on which we want to schedule the fragments (endpointPool).
3. Assign one fragment to each of the nodes from the above endpointPool to make sure minimum of one is assigned to each of them.
4. Calculate how many fragments to assign to each of the nodes in endpointPool based on how much work they have to do i.e. targetAllocation (proportional to localWorkUnits assigned to the node).
5. Go through the endpointPool in a round robin way and keep assigning fragments to individual nodes till their target allocation or maxWidthPerNodes is reached.
6. Stop when overall allocation reaches the total target i.e. width above. 
7. It is possible that some rowGroups  have preferred endPoints null (because there is no drill bit running on the hosts which have data for the rowGroup). In that case, we will have unassigned work Items.
8. Allocate the fragments for unassigned work items to active end points, making sure maxWidthPerNode constraint is honored.

Assignment(AssignmentCreator.getMappings):
1. When the system option parquet.use_local_affinity is set to true, assign each rowGroup to a fragment (round robin) on it’s preferredEndPoint (assignLocal).
2. If the preferredEndpoint is null or fragment is not available on that node, add it to unassignedList
3. Fallback to current algorithm to assign unassigned list of rowGroups from 2.



> Fragment planning causes Drillbits to read remote chunks when local copies are available
> ----------------------------------------------------------------------------------------
>
>                 Key: DRILL-4706
>                 URL: https://issues.apache.org/jira/browse/DRILL-4706
>             Project: Apache Drill
>          Issue Type: Bug
>          Components: Query Planning & Optimization
>    Affects Versions: 1.6.0
>         Environment: CentOS, RHEL
>            Reporter: Kunal Khatua
>            Assignee: Padma Penumarthy
>              Labels: performance, planning
>
> When a table (datasize=70GB) of 160 parquet files (each having a single rowgroup and fitting within one chunk) is available on a 10-node setup with replication=3 ; a pure data scan query causes about 2% of the data to be read remotely. 
> Even with the creation of metadata cache, the planner is selecting a sub-optimal plan of executing the SCAN fragments such that some of the data is served from a remote server. 



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)