[PIG-5029] Optimize sort case when data is skewed - ASF JIRA

Add vote

Voters

Watch issue

Watchers

Link

Clone

Update Comment Author

Replace String in Comment

Update Comment Visibility

Delete Comments

XML

Word

Printable

JSON

Details

Type: Sub-task
Status: Patch Available
Priority: Major
Resolution: Unresolved
Affects Version/s: None
Fix Version/s: spark-branch
Component/s: spark
Labels:
None

Description

In PigMix L9.pig

register $PIGMIX_JAR
A = load '$HDFS_ROOT/page_views' using org.apache.pig.test.pigmix.udf.PigPerformanceLoader()
    as (user, action, timespent, query_term, ip_addr, timestamp,
        estimated_revenue, page_info, page_links);
B = order A by query_term parallel $PARALLEL;
store B into '$PIGMIX_OUTPUT/L9out';

The pig physical plan will be changed to spark plan and to spark lineage:

[main] 2016-09-08 01:49:09,844 DEBUG converter.StoreConverter (StoreConverter.java:convert(110)) - RDD lineage: (23) MapPartitionsRDD[8] at map at StoreConverter.java:80 []
 |   MapPartitionsRDD[7] at mapPartitions at SortConverter.java:58 []
 |   ShuffledRDD[6] at sortByKey at SortConverter.java:56 []
 +-(23) MapPartitionsRDD[3] at map at SortConverter.java:49 []
    |   MapPartitionsRDD[2] at mapPartitions at ForEachConverter.java:64 []
    |   MapPartitionsRDD[1] at map at LoadConverter.java:127 []
    |   NewHadoopRDD[0] at newAPIHadoopRDD at LoadConverter.java:102 []

We use sortByKey to implement the sort feature. Although RangePartitioner is used by RDD.sortByKey and RangePartitiner will sample data and ranges the key roughly into equal range, the test result(attached document) shows that one partition will load most keys and take long time to finish.