[jira] [Updated] (TEZ-2496) Consider scheduling tasks in ShuffleVertexManager based on the partition sizes from the source

["Rajesh Balamohan (JIRA)" <ji...@apache.org>]

     [ https://issues.apache.org/jira/browse/TEZ-2496?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]

Rajesh Balamohan updated TEZ-2496:
----------------------------------
    Attachment: TEZ-2496.4.patch

Addressing [~sseth]'s comments
>> In JavaDocs - replace destination with 'physical output'
- Fixed.

>> Figuring out the bucket (getBucket) would be easier with an Enumeration. maybe a helper to convert a size to an enum.
- Fixed. Introduced enum

>> Use 4 bits instead of 6. I'm guessing this is 6 to optimize for the merge operation
- 0 was specifically added so that we can get rid of empty partitions. Using 0,1,100,1000

>> ShuffleVertexManager: PendingTaskInfo.stats - rename to what the parameter represents.
- Fixed. Renamed to outputStats.

>> Does configureTargetMapping(); need to be invoked inside the conditional which updates parallelism ?
- Fixed.

>> Sorter stats reporting: If I'm reading this correctly - stats are reported for each spill ? Shouldn't this be once at the end only. Especially since there's no increment API for the stats being reported. Alternately, report combined stats for all spills up to the last report time.
- Fixed.  Reporting this at the end of sorter close().

Addressing [~bikassaha]'s comments

>> Will thread safety be an issue if a different partition writers are writing the their partition sizes?
- Fixed. Moved this within write lock

>> Since sizeInMb is used in getBucket() can the whole calculation wrt size be moved to getBucket() to simplify the code.
- Moved this to enum

>> The actual int value is not being used anywhere other than as an exists check, right? Then why not simply writeBoolean?
- Fixed.

>> Duplication of existing code?
- Moved this to common code & reusing it.

>> Does computing partition size and sorting have to be done every time a task is scheduled?
- No; Changed it to run only when the config flag is enabled. Quite possible that larger vertices choose not to publish the stats.

>> From the API pov, this is introducing redundancy with the existing setDataSize() that sets it globally. Do both need to be called now? 
- Marked it optional.  Based on "tez.runtime.report.partition.stats" flag, one of it would be called in ExternalSorter.  This flag will also allow to share partition stats in vertex in optional way.

>> Do we have some before after numbers that show memory usage in the AM, say for a 10Kx10K job.  Ideally, a new case could be added in TestMemoryWithEvents that covers this case of sending per partition stats so that we can repro easily.
- Added testIOStatisticsMemory() for 10K x 10K scatter-gather.

{noformat}
With Patch (with stats reporting):
TestMockDAGAppMaster.java:testIOStatisticsMemory
Time taken : 186063
##### Heap utilization statistics [MB] for testIOStatisticsMemory
##### Used Memory:487
##### Free Memory:214
##### Total Memory:702
##### Max Memory:3641

With Patch (no stats reporting)
Time taken : 178782
##### Heap utilization statistics [MB] for testIOStatisticsMemory
##### Used Memory:95
##### Free Memory:199
##### Total Memory:295
##### Max Memory:3641
{noformat}

Increase in used memory for 10K is mainly due to TaskStatistics storing BitSets (~250 MB for for storing 10K x 10K stats. Checked via yourkit profiler). Depending on memory usage, it should be possible for higher level vertices to stop publishing these stats as well. 

> Consider scheduling tasks in ShuffleVertexManager based on the partition sizes from the source
> ----------------------------------------------------------------------------------------------
>
>                 Key: TEZ-2496
>                 URL: https://issues.apache.org/jira/browse/TEZ-2496
>             Project: Apache Tez
>          Issue Type: Improvement
>            Reporter: Rajesh Balamohan
>            Assignee: Rajesh Balamohan
>         Attachments: TEZ-2496.1.patch, TEZ-2496.2.patch, TEZ-2496.3.patch, TEZ-2496.4.patch
>
>
> Consider scheduling tasks in ShuffleVertexManager based on the partition sizes from the source.  This would be helpful in scenarios, where there is limited resources (or concurrent jobs running or multiple waves) with dataskew and the task which gets large amount of data gets sceheduled much later.
> e.g Consider the following hive query running in a queue with limited capacity (42 slots in total) @ 200 GB scale
> {noformat}
> CREATE TEMPORARY TABLE sampleData AS
>   SELECT CASE
>            WHEN ss_sold_time_sk IS NULL THEN 70429
>            ELSE ss_sold_time_sk
>        END AS ss_sold_time_sk,
>        ss_item_sk,
>        ss_customer_sk,
>        ss_cdemo_sk,
>        ss_hdemo_sk,
>        ss_addr_sk,
>        ss_store_sk,
>        ss_promo_sk,
>        ss_ticket_number,
>        ss_quantity,
>        ss_wholesale_cost,
>        ss_list_price,
>        ss_sales_price,
>        ss_ext_discount_amt,
>        ss_ext_sales_price,
>        ss_ext_wholesale_cost,
>        ss_ext_list_price,
>        ss_ext_tax,
>        ss_coupon_amt,
>        ss_net_paid,
>        ss_net_paid_inc_tax,
>        ss_net_profit,
>        ss_sold_date_sk
>   FROM store_sales distribute by ss_sold_time_sk;
> {noformat}
> This generated 39 maps and 134 reduce slots (3 reduce waves). When lots of nulls are there for ss_sold_time_sk, it would tend to have data skew towards 70429.  If the reducer which gets this data gets scheduled much earlier (i.e in first wave itself), entire job would finish fast.



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