[
https://issues.apache.org/jira/browse/SPARK-31948?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]
zhengruifeng updated SPARK-31948:
---------------------------------
Description:
{{1, aggregateByKey,}} {{reduceByKey}} and {{foldByKey}} will always perform
{{mapSideCombine}};
However, this can be skiped sometime, specially in ML (RobustScaler):
{code:java}
vectors.mapPartitions { iter =>
if (iter.hasNext) {
val summaries = Array.fill(numFeatures)(
new QuantileSummaries(QuantileSummaries.defaultCompressThreshold,
relativeError))
while (iter.hasNext) {
val vec = iter.next
vec.foreach { (i, v) => if (!v.isNaN) summaries(i) =
summaries(i).insert(v) }
}
Iterator.tabulate(numFeatures)(i => (i, summaries(i).compress))
} else Iterator.empty
}.reduceByKey { case (s1, s2) => s1.merge(s2) } {code}
This {{reduceByKey}} in {{RobustScaler}} does not need {{mapSideCombine}} at
all, similar places exist in {{KMeans}}, {{GMM}}, etc;
to my knowledge, we do not need {{mapSideCombine}} if the reduction factor
isn't high;
2, {{treeAggregate}} and {{treeReduce}} are based on {{foldByKey}}, the
{{mapSideCombine in the first call of {{foldByKey}}{{}}}}{{ can also be
avoided.}}
{{SPARK-772: "Map side combine in group by key case does not reduce the amount
of data shuffled. Instead, it forces a lot more objects to go into old gen, and
leads to worse GC."}}
{{So what about:}}
{{1, exposing mapSideCombine in aggByKey/reduceByKey/foldByKey, so that user
can disable unnecessary }}{{mapSideCombine}}{{;}}
{{2, disabling the }}\{{mapSideCombine in the first call of }}\{{foldByKey in
}}{{treeAggregate}}{\{ and }}{{treeReduce}}{{;}}{{}}
{{3, }}\{{disabling the uncessary }}{{mapSideCombine in ML;}}{{}}
{{friendly ping [~srowen] [~huaxingao] [~weichenxu123] [~hyukjin.kwon]
[~viirya] }}
was:
{{1, aggregateByKey,}} {{reduceByKey}} and {{foldByKey}} will always perform
{{mapSideCombine}};
However, this can be skiped sometime, specially in ML (RobustScaler):
{code:java}
vectors.mapPartitions { iter =>
if (iter.hasNext) {
val summaries = Array.fill(numFeatures)(
new QuantileSummaries(QuantileSummaries.defaultCompressThreshold,
relativeError))
while (iter.hasNext) {
val vec = iter.next
vec.foreach { (i, v) => if (!v.isNaN) summaries(i) =
summaries(i).insert(v) }
}
Iterator.tabulate(numFeatures)(i => (i, summaries(i).compress))
} else Iterator.empty
}.reduceByKey { case (s1, s2) => s1.merge(s2) } {code}
This {{reduceByKey}} in {{RobustScaler}} does not need {{mapSideCombine}} at
all, similar places exist in {{KMeans}}, {{GMM}}, etc;
to my knowledge, we do not need {{mapSideCombine}} if the reduction factor
isn't high;
2, {{treeAggregate}} and {{treeReduce}} are based on {{foldByKey}}, the
{{mapSideCombine in the first call of }}{{foldByKey can also be avoided.}}{{}}
{{SPARK-772: "Map side combine in group by key case does not reduce the amount
of data shuffled. Instead, it forces a lot more objects to go into old gen, and
leads to worse GC."}}
{{So what about:}}
{{1, exposing mapSideCombine in aggByKey/reduceByKey/foldByKey, so that user
can disable unnecessary }}{{mapSideCombine}}{{;}}
{{2, disabling the }}{{mapSideCombine in the first call of }}{{foldByKey in
}}{{treeAggregate}}{{ and }}{{treeReduce}}{{;}}{{}}
{{3, }}{{disabling the uncessary }}{{mapSideCombine in ML;}}{{}}
{{friendly ping [~srowen] [~huaxingao] [~weichenxu123] [~hyukjin.kwon]
[~viirya] }}
> expose mapSideCombine in aggByKey/reduceByKey/foldByKey
> -------------------------------------------------------
>
> Key: SPARK-31948
> URL: https://issues.apache.org/jira/browse/SPARK-31948
> Project: Spark
> Issue Type: Improvement
> Components: ML, Spark Core
> Affects Versions: 3.1.0
> Reporter: zhengruifeng
> Priority: Minor
>
> {{1, aggregateByKey,}} {{reduceByKey}} and {{foldByKey}} will always perform
> {{mapSideCombine}};
> However, this can be skiped sometime, specially in ML (RobustScaler):
> {code:java}
> vectors.mapPartitions { iter =>
> if (iter.hasNext) {
> val summaries = Array.fill(numFeatures)(
> new QuantileSummaries(QuantileSummaries.defaultCompressThreshold,
> relativeError))
> while (iter.hasNext) {
> val vec = iter.next
> vec.foreach { (i, v) => if (!v.isNaN) summaries(i) =
> summaries(i).insert(v) }
> }
> Iterator.tabulate(numFeatures)(i => (i, summaries(i).compress))
> } else Iterator.empty
> }.reduceByKey { case (s1, s2) => s1.merge(s2) } {code}
>
> This {{reduceByKey}} in {{RobustScaler}} does not need {{mapSideCombine}} at
> all, similar places exist in {{KMeans}}, {{GMM}}, etc;
> to my knowledge, we do not need {{mapSideCombine}} if the reduction factor
> isn't high;
>
> 2, {{treeAggregate}} and {{treeReduce}} are based on {{foldByKey}}, the
> {{mapSideCombine in the first call of {{foldByKey}}{{}}}}{{ can also be
> avoided.}}
>
> {{SPARK-772: "Map side combine in group by key case does not reduce the
> amount of data shuffled. Instead, it forces a lot more objects to go into old
> gen, and leads to worse GC."}}
>
> {{So what about:}}
> {{1, exposing mapSideCombine in aggByKey/reduceByKey/foldByKey, so that user
> can disable unnecessary }}{{mapSideCombine}}{{;}}
> {{2, disabling the }}\{{mapSideCombine in the first call of }}\{{foldByKey in
> }}{{treeAggregate}}{\{ and }}{{treeReduce}}{{;}}{{}}
> {{3, }}\{{disabling the uncessary }}{{mapSideCombine in ML;}}{{}}
>
>
> {{friendly ping [~srowen] [~huaxingao] [~weichenxu123] [~hyukjin.kwon]
> [~viirya] }}
>
>
>
>
>
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org
For additional commands, e-mail: issues-h...@spark.apache.org
