[ https://issues.apache.org/jira/browse/SPARK-27940?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel ]
Josh Rosen updated SPARK-27940: ------------------------------- Description: {{SubtractedRDD}}, which is used to implement {{RDD.subtract()}} and {{PairRDDFunctions.subtractByKey()}}, currently buffers one partition in memory and does not support spilling: [https://github.com/apache/spark/blob/v2.4.3/core/src/main/scala/org/apache/spark/rdd/SubtractedRDD.scala#L42] In principle, we could implement {{subtractByKey}} as a left-outer join followed by a filter (e.g. as an antijoin), but the Scaladoc explains why this approach wasn't taken: {code:java} * It is possible to implement this operation with just `cogroup`, but * that is less efficient because all of the entries from `rdd2`, for * both matching and non-matching values in `rdd1`, are kept in the * JHashMap until the end.{code} For example, if we have {{left.subtractByKey(right)}} and {{right}} has hundreds of occurrences of a key then we'd end up buffering hundreds of tuples. Instead, maybe we could implement a sort-merge join where we build an {{ExternalAppendOnlyMap}} of unique {{right}} keys, use an {{ExternalSorter}} to sort the {{left}}| input, then iterate over both sorted iterators and perform a merge. Note that this problem only impacts the RDD API. Here are some existing workarounds for this OOM-proneness: * Use more partitions: e.g. {{left.subtractByKey(right, 2000)}} (or pass in a custom partitioner). This may not help if you have heavily skewed keys, though. * Use a left join followed by filter: {code:java} left .leftOuterJoin(right) .filter(!_._2._2.isDefined) .mapValues(_._1){code} If you wanted to further optimize, you could replace {{right}} values with dummy placeholders to avoid having to shuffle them: {code:java} left .leftOuterJoin(right.map { case (k, v) => (k, 0) }) .filter(!_._2._2.isDefined) .mapValues(_._1){code} * Use DataFrames / Datasets instead of RDDs. was: {{SubtractedRDD}}, which is used to implement {{RDD.subtract()}} and {{PairRDDFunctions.subtractByKey()}}, currently buffers one partition in memory and does not support spilling: [https://github.com/apache/spark/blob/v2.4.3/core/src/main/scala/org/apache/spark/rdd/SubtractedRDD.scala#L42] In principle, we could implement {{subtractByKey}} as a left-outer join followed by a filter (e.g. as an antijoin), but the Scaladoc explains why this approach wasn't taken: {code:java} * It is possible to implement this operation with just `cogroup`, but * that is less efficient because all of the entries from `rdd2`, for * both matching and non-matching values in `rdd1`, are kept in the * JHashMap until the end.{code} For example, if we have {{left.subtractByKey(right)}} and {{right}} has hundreds of occurrences of a key then we'd end up buffering hundreds of tuples. Instead, maybe we could implement a sort-merge join where we build an {{ExternalAppendOnlyMap}} of unique {{right}} keys, use an {{ExternalSorter}} to sort the {{left}}| input, then iterate over both sorted iterators and perform a merge. Note that this problem only impacts the RDD API. Here are some existing workarounds for this OOM-proneness: * Use more partitions: e.g. {{left.subtractByKey(right, 2000)}} (or pass in a custom partitioner). This may not help if you have heavily skewed keys, though. * Use a left join followed by filter: {code:java} left .leftOuterJoin(right) .collect { case (k, (lv, None)) => (k, lv) }{code} If you wanted to further optimize, you could replace {{right}} values with dummy placeholders to avoid having to shuffle them: {code:java} left .leftOuterJoin(right.map { case (k, v) => (k, 0) }) .collect { case (k, (lv, None)) => (k, lv) }{code} * Use DataFrames / Datasets instead of RDDs. > SubtractedRDD is OOM-prone because it does not support spilling > --------------------------------------------------------------- > > Key: SPARK-27940 > URL: https://issues.apache.org/jira/browse/SPARK-27940 > Project: Spark > Issue Type: Bug > Components: Spark Core > Affects Versions: 2.4.0 > Reporter: Josh Rosen > Priority: Minor > > {{SubtractedRDD}}, which is used to implement {{RDD.subtract()}} and > {{PairRDDFunctions.subtractByKey()}}, currently buffers one partition in > memory and does not support spilling: > [https://github.com/apache/spark/blob/v2.4.3/core/src/main/scala/org/apache/spark/rdd/SubtractedRDD.scala#L42] > In principle, we could implement {{subtractByKey}} as a left-outer join > followed by a filter (e.g. as an antijoin), but the Scaladoc explains why > this approach wasn't taken: > {code:java} > * It is possible to implement this operation with just `cogroup`, but > * that is less efficient because all of the entries from `rdd2`, for > * both matching and non-matching values in `rdd1`, are kept in the > * JHashMap until the end.{code} > For example, if we have {{left.subtractByKey(right)}} and {{right}} has > hundreds of occurrences of a key then we'd end up buffering hundreds of > tuples. > Instead, maybe we could implement a sort-merge join where we build an > {{ExternalAppendOnlyMap}} of unique {{right}} keys, use an {{ExternalSorter}} > to sort the {{left}}| input, then iterate over both sorted iterators and > perform a merge. > Note that this problem only impacts the RDD API. > Here are some existing workarounds for this OOM-proneness: > * Use more partitions: e.g. {{left.subtractByKey(right, 2000)}} (or pass in > a custom partitioner). This may not help if you have heavily skewed keys, > though. > * Use a left join followed by filter: > {code:java} > left > .leftOuterJoin(right) > .filter(!_._2._2.isDefined) > .mapValues(_._1){code} > If you wanted to further optimize, you could replace {{right}} values with > dummy placeholders to avoid having to shuffle them: > {code:java} > left > .leftOuterJoin(right.map { case (k, v) => (k, 0) }) > .filter(!_._2._2.isDefined) > .mapValues(_._1){code} > * Use DataFrames / Datasets instead of RDDs. -- This message was sent by Atlassian JIRA (v7.6.3#76005) --------------------------------------------------------------------- To unsubscribe, e-mail: issues-unsubscr...@spark.apache.org For additional commands, e-mail: issues-h...@spark.apache.org