Uploaded image for project: 'Spark'
  1. Spark
  2. SPARK-27940

SubtractedRDD is OOM-prone because it does not support spilling

    XMLWordPrintableJSON

    Details

    • Type: Bug
    • Status: Open
    • Priority: Minor
    • Resolution: Unresolved
    • Affects Version/s: 2.4.0
    • Fix Version/s: None
    • Component/s: Spark Core
    • Labels:
      None

      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:

      * 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.

      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: 
        left
          .leftOuterJoin(right)
          .filter(!_._2._2.isDefined)
          .mapValues(_._1)

        If you wanted to further optimize, you could replace right values with dummy placeholders to avoid having to shuffle them:

        left
          .leftOuterJoin(right.map { case (k, v) => (k, 0) })
          .filter(!_._2._2.isDefined)
          .mapValues(_._1)
      • Use DataFrames / Datasets instead of RDDs.

        Attachments

          Activity

            People

            • Assignee:
              Unassigned
              Reporter:
              joshrosen Josh Rosen
            • Votes:
              0 Vote for this issue
              Watchers:
              3 Start watching this issue

              Dates

              • Created:
                Updated: