spark full outer join 数据倾斜导致OOM

doExecute 
   ||
   \/
case FullOuter =>
     val leftNullRow = new GenericInternalRow(left.output.length)
     val rightNullRow = new GenericInternalRow(right.output.length)
     val smjScanner = new SortMergeFullOuterJoinScanner(
     leftKeyGenerator = createLeftKeyGenerator(),
     rightKeyGenerator = createRightKeyGenerator(),
     keyOrdering,
     leftIter = RowIterator.fromScala(leftIter),
     rightIter = RowIterator.fromScala(rightIter),
     boundCondition,
     leftNullRow,
     rightNullRow)
     new FullOuterIterator(
      smjScanner,
      resultProj,
      numOutputRows).toScala

private def findMatchingRows(matchingKey: InternalRow): Unit = {
    leftMatches.clear()
    rightMatches.clear()
    leftIndex = 0
    rightIndex = 0
    while (leftRowKey != null && keyOrdering.compare(leftRowKey, matchingKey) == 0) {
      leftMatches += leftRow.copy()
      advancedLeft()
    }
    while (rightRowKey != null && keyOrdering.compare(rightRowKey, matchingKey) == 0) {
      rightMatches += rightRow.copy()
      advancedRight()
    }
    if (leftMatches.size <= leftMatched.capacity) {
      leftMatched.clearUntil(leftMatches.size)
    } else {
      leftMatched = new BitSet(leftMatches.size)
    }
    if (rightMatches.size <= rightMatched.capacity) {
      rightMatched.clearUntil(rightMatches.size)
    } else {
      rightMatched = new BitSet(rightMatches.size)
    }
  }