diff --git a/core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriter.java b/core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriter.java
index fdb309e365f69..32bca0a1feb48 100644
--- a/core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriter.java
+++ b/core/src/main/java/org/apache/spark/shuffle/unsafe/UnsafeShuffleWriter.java
@@ -48,7 +48,7 @@
 import org.apache.spark.serializer.SerializationStream;
 import org.apache.spark.serializer.Serializer;
 import org.apache.spark.serializer.SerializerInstance;
-import org.apache.spark.shuffle.IndexShuffleBlockResolver;
+import org.apache.spark.shuffle.BinaryShuffleWriter;
 import org.apache.spark.shuffle.ShuffleMemoryManager;
 import org.apache.spark.shuffle.ShuffleWriter;
 import org.apache.spark.storage.BlockManager;
@@ -67,14 +67,12 @@ public class UnsafeShuffleWriter<K, V> extends ShuffleWriter<K, V> {
   static final int INITIAL_SORT_BUFFER_SIZE = 4096;
 
   private final BlockManager blockManager;
-  private final IndexShuffleBlockResolver shuffleBlockResolver;
+  private final BinaryShuffleWriter binaryShuffleWriter;
   private final TaskMemoryManager memoryManager;
   private final ShuffleMemoryManager shuffleMemoryManager;
   private final SerializerInstance serializer;
   private final Partitioner partitioner;
   private final ShuffleWriteMetrics writeMetrics;
-  private final int shuffleId;
-  private final int mapId;
   private final TaskContext taskContext;
   private final SparkConf sparkConf;
   private final boolean transferToEnabled;
@@ -101,28 +99,25 @@ private static final class MyByteArrayOutputStream extends ByteArrayOutputStream
 
   public UnsafeShuffleWriter(
       BlockManager blockManager,
-      IndexShuffleBlockResolver shuffleBlockResolver,
+      BinaryShuffleWriter binaryShuffleWriter,
       TaskMemoryManager memoryManager,
       ShuffleMemoryManager shuffleMemoryManager,
-      UnsafeShuffleHandle<K, V> handle,
-      int mapId,
+      Partitioner partitioner,
+      Option<Serializer> serializer,
       TaskContext taskContext,
       SparkConf sparkConf) throws IOException {
-    final int numPartitions = handle.dependency().partitioner().numPartitions();
+    final int numPartitions = partitioner.numPartitions();
     if (numPartitions > UnsafeShuffleManager.MAX_SHUFFLE_OUTPUT_PARTITIONS()) {
       throw new IllegalArgumentException(
         "UnsafeShuffleWriter can only be used for shuffles with at most " +
           UnsafeShuffleManager.MAX_SHUFFLE_OUTPUT_PARTITIONS() + " reduce partitions");
     }
     this.blockManager = blockManager;
-    this.shuffleBlockResolver = shuffleBlockResolver;
+    this.binaryShuffleWriter = binaryShuffleWriter;
     this.memoryManager = memoryManager;
     this.shuffleMemoryManager = shuffleMemoryManager;
-    this.mapId = mapId;
-    final ShuffleDependency<K, V, V> dep = handle.dependency();
-    this.shuffleId = dep.shuffleId();
-    this.serializer = Serializer.getSerializer(dep.serializer()).newInstance();
-    this.partitioner = dep.partitioner();
+    this.serializer = Serializer.getSerializer(serializer).newInstance();
+    this.partitioner = partitioner;
     this.writeMetrics = new ShuffleWriteMetrics();
     taskContext.taskMetrics().shuffleWriteMetrics_$eq(Option.apply(writeMetrics));
     this.taskContext = taskContext;
@@ -226,7 +221,7 @@ void closeAndWriteOutput() throws IOException {
         }
       }
     }
-    shuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths);
+    binaryShuffleWriter.commit(partitionLengths);
     mapStatus = MapStatus$.MODULE$.apply(blockManager.shuffleServerId(), partitionLengths);
   }
 
@@ -260,7 +255,7 @@ void forceSorterToSpill() throws IOException {
    * @return the partition lengths in the merged file.
    */
   private long[] mergeSpills(SpillInfo[] spills) throws IOException {
-    final File outputFile = shuffleBlockResolver.getDataFile(shuffleId, mapId);
+    final File outputFile = binaryShuffleWriter.getDataFile();
     final boolean compressionEnabled = sparkConf.getBoolean("spark.shuffle.compress", true);
     final CompressionCodec compressionCodec = CompressionCodec$.MODULE$.createCodec(sparkConf);
     final boolean fastMergeEnabled =
@@ -474,7 +469,7 @@ public Option<MapStatus> stop(boolean success) {
           return Option.apply(mapStatus);
         } else {
           // The map task failed, so delete our output data.
-          shuffleBlockResolver.removeDataByMap(shuffleId, mapId);
+          binaryShuffleWriter.abort();
           return Option.apply(null);
         }
       }
diff --git a/core/src/main/scala/org/apache/spark/ContextCleaner.scala b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
index d23c1533db758..7dbed6b2ac8f1 100644
--- a/core/src/main/scala/org/apache/spark/ContextCleaner.scala
+++ b/core/src/main/scala/org/apache/spark/ContextCleaner.scala
@@ -131,7 +131,7 @@ private[spark] class ContextCleaner(sc: SparkContext) extends Logging {
   }
 
   /** Register a ShuffleDependency for cleanup when it is garbage collected. */
-  def registerShuffleForCleanup(shuffleDependency: ShuffleDependency[_, _, _]): Unit = {
+  def registerShuffleForCleanup(shuffleDependency: BaseShuffleDependency[_]): Unit = {
     registerForCleanup(shuffleDependency, CleanShuffle(shuffleDependency.shuffleId))
   }
 
diff --git a/core/src/main/scala/org/apache/spark/Dependency.scala b/core/src/main/scala/org/apache/spark/Dependency.scala
index 9aafc9eb1cde7..40bbc2211fea1 100644
--- a/core/src/main/scala/org/apache/spark/Dependency.scala
+++ b/core/src/main/scala/org/apache/spark/Dependency.scala
@@ -22,7 +22,7 @@ import scala.reflect.ClassTag
 import org.apache.spark.annotation.DeveloperApi
 import org.apache.spark.rdd.RDD
 import org.apache.spark.serializer.Serializer
-import org.apache.spark.shuffle.ShuffleHandle
+import org.apache.spark.shuffle.{BinaryShuffleWriter, ShuffleHandle}
 
 /**
  * :: DeveloperApi ::
@@ -68,15 +68,14 @@ abstract class NarrowDependency[T](_rdd: RDD[T]) extends Dependency[T] {
  */
 @DeveloperApi
 class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
-    @transient private val _rdd: RDD[_ <: Product2[K, V]],
+    _rdd: RDD[_ <: Product2[K, V]],
     val partitioner: Partitioner,
     val serializer: Option[Serializer] = None,
     val keyOrdering: Option[Ordering[K]] = None,
     val aggregator: Option[Aggregator[K, V, C]] = None,
     val mapSideCombine: Boolean = false)
-  extends Dependency[Product2[K, V]] {
-
-  override def rdd: RDD[Product2[K, V]] = _rdd.asInstanceOf[RDD[Product2[K, V]]]
+  extends BaseShuffleDependency[Product2[K, V]](
+    _rdd.asInstanceOf[RDD[Product2[K, V]]], partitioner.numPartitions) {
 
   private[spark] val keyClassName: String = reflect.classTag[K].runtimeClass.getName
   private[spark] val valueClassName: String = reflect.classTag[V].runtimeClass.getName
@@ -85,14 +84,28 @@ class ShuffleDependency[K: ClassTag, V: ClassTag, C: ClassTag](
   private[spark] val combinerClassName: Option[String] =
     Option(reflect.classTag[C]).map(_.runtimeClass.getName)
 
-  val shuffleId: Int = _rdd.context.newShuffleId()
-
   val shuffleHandle: ShuffleHandle = _rdd.context.env.shuffleManager.registerShuffle(
     shuffleId, _rdd.partitions.size, this)
+}
+
+
+private[spark] abstract class BaseShuffleDependency[T: ClassTag](
+    @transient private val _rdd: RDD[T],
+    val numPartitions: Int) extends Dependency[T] {
+
+  override def rdd: RDD[T] = _rdd
+
+  val shuffleId: Int = _rdd.context.newShuffleId()
 
   _rdd.sparkContext.cleaner.foreach(_.registerShuffleForCleanup(this))
 }
 
+private[spark] class BinaryShuffleDependency[T: ClassTag](
+    _rdd: RDD[T],
+    numPartitions: Int,
+    val writeFunc: (TaskContext, BinaryShuffleWriter, Iterator[T]) => Unit)
+  extends BaseShuffleDependency[T](_rdd, numPartitions)
+
 
 /**
  * :: DeveloperApi ::
diff --git a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
index 94eb8daa85c53..2a9ecae44172b 100644
--- a/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
+++ b/core/src/main/scala/org/apache/spark/MapOutputTracker.scala
@@ -145,11 +145,11 @@ private[spark] abstract class MapOutputTracker(conf: SparkConf) extends Logging
   /**
    * Return statistics about all of the outputs for a given shuffle.
    */
-  def getStatistics(dep: ShuffleDependency[_, _, _]): MapOutputStatistics = {
+  def getStatistics(dep: BaseShuffleDependency[_]): MapOutputStatistics = {
     val statuses = getStatuses(dep.shuffleId)
     // Synchronize on the returned array because, on the driver, it gets mutated in place
     statuses.synchronized {
-      val totalSizes = new Array[Long](dep.partitioner.numPartitions)
+      val totalSizes = new Array[Long](dep.numPartitions)
       for (s <- statuses) {
         for (i <- 0 until totalSizes.length) {
           totalSizes(i) += s.getSizeForBlock(i)
diff --git a/core/src/main/scala/org/apache/spark/rdd/BinaryShuffledRDD.scala b/core/src/main/scala/org/apache/spark/rdd/BinaryShuffledRDD.scala
new file mode 100644
index 0000000000000..1e47af78a8dbc
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/rdd/BinaryShuffledRDD.scala
@@ -0,0 +1,115 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.rdd
+
+import java.io.InputStream
+
+import scala.reflect.ClassTag
+
+import org.apache.spark.{BinaryShuffleDependency, Dependency, SparkEnv, TaskContext, Partition}
+import org.apache.spark.shuffle.BinaryShuffleWriter
+import org.apache.spark.storage.ShuffleBlockFetcherIterator
+
+/**
+ * This class allows non-core Spark components to perform shuffles with customized data writing
+ * and reading. It effectively allows non-core components to customize all aspects of shuffle except
+ * for the actual network transfer.
+ *
+ * Compared to the existing shuffle APIs, such as [[ShuffledRDD]] and
+ * [[org.apache.spark.ShuffleDependency]], this API is lower-level and thus more general:
+ *
+ *   - The shuffle write and read paths are not record-oriented and are not coupled to Scala tuples
+ *     or key-value pairs. This allows custom shuffle implementations to operate on batches of
+ *     records or to write shuffle data in a column-oriented format.
+ *   - This API says nothing about serialization, sorting, aggregation, or partitioning; instead,
+ *     it only captures the structure of the wide dependency and exposes a low-level interface for
+ *     writing opaque binary shuffle data.
+ *   - The write interfaces only expose file names and success / abort methods. As a result, the
+ *     actual shuffle writing may be performed by an external process or by native (non-JVM) code.
+ *
+ * Note that this is _not_ intended to be a DeveloperApi yet and thus should not be used by code
+ * outside of Spark itself. While we may eventually stabilize a version of this API, the current
+ * API is not subject to any binary compatibility guarantees and thus should not be used by
+ * third-party code.
+ *
+ * @param prev the RDD being shuffled.
+ * @param numPartitions the number of partitions that will result from the shuffle.
+ * @tparam ParentType the type of records in the RDD being shuffled.
+ * @tparam OutputType the type of records in the post-shuffle RDD. Note that this may be different
+ *                    than [[ParentType]] if the custom shuffle is performing aggregation or
+ *                    grouping.
+ */
+private[spark] abstract class BinaryShuffledRDD[ParentType: ClassTag, OutputType: ClassTag](
+    @transient var prev: RDD[ParentType],
+    numPartitions: Int
+  ) extends RDD[OutputType](prev.context, Nil) {
+
+  // -- Methods to be implemented by subclasses ---------------------------------------------------
+
+  /**
+   * Called on the map side to write shuffle data.
+   *
+   * Note that it is the implementor's responsibility to perform any desired compression of the
+   * shuffle data.
+   *
+   * @param context the TaskContext.
+   * @param writer a handle which exposes the path where shuffle data should be written.
+   *               See the documentation in [[BinaryShuffleWriter]] for a description of the
+   *               expected file format.
+   * @param iter an iterator of records from the RDD being shuffled.
+   */
+  def write(context: TaskContext, writer: BinaryShuffleWriter, iter: Iterator[ParentType]): Unit
+
+  /**
+   * Called on the reduce side to read shuffle data.
+   *
+   * @param context the TaskContext.
+   * @param iter an iterator of input streams, one per map partition. These streams correspond,
+   *             verbatim, to the per-reducer data written on the map side.
+   * @return an iterator of records that were decoded from the input streams.
+   */
+  def read(context: TaskContext, iter: Iterator[InputStream]): Iterator[OutputType]
+
+  // -- Private internal methods ------------------------------------------------------------------
+
+  final override def getPartitions: Array[Partition] = {
+    Array.tabulate[Partition](numPartitions)(i => new ShuffledRDDPartition(i))
+  }
+
+  final override def getDependencies: Seq[Dependency[_]] = {
+    List(new BinaryShuffleDependency[ParentType](prev, numPartitions, write))
+  }
+
+  final override def compute(split: Partition, context: TaskContext): Iterator[OutputType] = {
+    val dep = dependencies.head.asInstanceOf[BinaryShuffleDependency[ParentType]]
+    val blockManager = SparkEnv.get.blockManager
+    val blockFetcherItr = new ShuffleBlockFetcherIterator(
+      context,
+      blockManager.shuffleClient,
+      blockManager,
+      SparkEnv.get.mapOutputTracker.getMapSizesByExecutorId(dep.shuffleId, split.index),
+      // Note: we use getSizeAsMb when no suffix is provided for backwards compatibility
+      SparkEnv.get.conf.getSizeAsMb("spark.reducer.maxSizeInFlight", "48m") * 1024 * 1024)
+    read(context, blockFetcherItr.map(_._2))
+  }
+
+  override def clearDependencies() {
+    super.clearDependencies()
+    prev = null
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
index 394228b2728d1..ecd3d0d2854bd 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/DAGScheduler.scala
@@ -292,7 +292,7 @@ class DAGScheduler(
    * Get or create a shuffle map stage for the given shuffle dependency's map side.
    */
   private def getShuffleMapStage(
-      shuffleDep: ShuffleDependency[_, _, _],
+      shuffleDep: BaseShuffleDependency[_],
       firstJobId: Int): ShuffleMapStage = {
     shuffleToMapStage.get(shuffleDep.shuffleId) match {
       case Some(stage) => stage
@@ -326,7 +326,7 @@ class DAGScheduler(
   private def newShuffleMapStage(
       rdd: RDD[_],
       numTasks: Int,
-      shuffleDep: ShuffleDependency[_, _, _],
+      shuffleDep: BaseShuffleDependency[_],
       firstJobId: Int,
       callSite: CallSite): ShuffleMapStage = {
     val (parentStages: List[Stage], id: Int) = getParentStagesAndId(rdd, firstJobId)
@@ -361,7 +361,7 @@ class DAGScheduler(
    * recovered from the MapOutputTracker
    */
   private def newOrUsedShuffleStage(
-      shuffleDep: ShuffleDependency[_, _, _],
+      shuffleDep: BaseShuffleDependency[_],
       firstJobId: Int): ShuffleMapStage = {
     val rdd = shuffleDep.rdd
     val numTasks = rdd.partitions.length
@@ -399,7 +399,7 @@ class DAGScheduler(
         // we can't do it in its constructor because # of partitions is unknown
         for (dep <- r.dependencies) {
           dep match {
-            case shufDep: ShuffleDependency[_, _, _] =>
+            case shufDep: BaseShuffleDependency[_] =>
               parents += getShuffleMapStage(shufDep, firstJobId)
             case _ =>
               waitingForVisit.push(dep.rdd)
@@ -415,8 +415,8 @@ class DAGScheduler(
   }
 
   /** Find ancestor shuffle dependencies that are not registered in shuffleToMapStage yet */
-  private def getAncestorShuffleDependencies(rdd: RDD[_]): Stack[ShuffleDependency[_, _, _]] = {
-    val parents = new Stack[ShuffleDependency[_, _, _]]
+  private def getAncestorShuffleDependencies(rdd: RDD[_]): Stack[BaseShuffleDependency[_]] = {
+    val parents = new Stack[BaseShuffleDependency[_]]
     val visited = new HashSet[RDD[_]]
     // We are manually maintaining a stack here to prevent StackOverflowError
     // caused by recursively visiting
@@ -426,7 +426,7 @@ class DAGScheduler(
         visited += r
         for (dep <- r.dependencies) {
           dep match {
-            case shufDep: ShuffleDependency[_, _, _] =>
+            case shufDep: BaseShuffleDependency[_] =>
               if (!shuffleToMapStage.contains(shufDep.shuffleId)) {
                 parents.push(shufDep)
               }
@@ -457,7 +457,7 @@ class DAGScheduler(
         if (rddHasUncachedPartitions) {
           for (dep <- rdd.dependencies) {
             dep match {
-              case shufDep: ShuffleDependency[_, _, _] =>
+              case shufDep: BaseShuffleDependency[_] =>
                 val mapStage = getShuffleMapStage(shufDep, stage.firstJobId)
                 if (!mapStage.isAvailable) {
                   missing += mapStage
@@ -873,7 +873,7 @@ class DAGScheduler(
   }
 
   private[scheduler] def handleMapStageSubmitted(jobId: Int,
-      dependency: ShuffleDependency[_, _, _],
+      dependency: BaseShuffleDependency[_],
       callSite: CallSite,
       listener: JobListener,
       properties: Properties) {
@@ -1499,7 +1499,7 @@ class DAGScheduler(
         visitedRdds += rdd
         for (dep <- rdd.dependencies) {
           dep match {
-            case shufDep: ShuffleDependency[_, _, _] =>
+            case shufDep: BaseShuffleDependency[_] =>
               val mapStage = getShuffleMapStage(shufDep, stage.firstJobId)
               if (!mapStage.isAvailable) {
                 waitingForVisit.push(mapStage.rdd)
@@ -1577,7 +1577,7 @@ class DAGScheduler(
     // have at least REDUCER_PREF_LOCS_FRACTION of data as preferred locations
     if (shuffleLocalityEnabled && rdd.partitions.length < SHUFFLE_PREF_REDUCE_THRESHOLD) {
       rdd.dependencies.foreach {
-        case s: ShuffleDependency[_, _, _] =>
+        case s: BaseShuffleDependency[_] =>
           if (s.rdd.partitions.length < SHUFFLE_PREF_MAP_THRESHOLD) {
             // Get the preferred map output locations for this reducer
             val topLocsForReducer = mapOutputTracker.getLocationsWithLargestOutputs(s.shuffleId,
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala b/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala
index 7d92960876403..425540282cadb 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapStage.scala
@@ -17,7 +17,7 @@
 
 package org.apache.spark.scheduler
 
-import org.apache.spark.ShuffleDependency
+import org.apache.spark.{BaseShuffleDependency, ShuffleDependency}
 import org.apache.spark.rdd.RDD
 import org.apache.spark.storage.BlockManagerId
 import org.apache.spark.util.CallSite
@@ -40,7 +40,7 @@ private[spark] class ShuffleMapStage(
     parents: List[Stage],
     firstJobId: Int,
     callSite: CallSite,
-    val shuffleDep: ShuffleDependency[_, _, _])
+    val shuffleDep: BaseShuffleDependency[_])
   extends Stage(id, rdd, numTasks, parents, firstJobId, callSite) {
 
   override def toString: String = "ShuffleMapStage " + id
diff --git a/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala b/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala
index f478f9982afef..8052ce7ef50f3 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/ShuffleMapTask.scala
@@ -20,11 +20,12 @@ package org.apache.spark.scheduler
 import java.nio.ByteBuffer
 
 import scala.language.existentials
+import scala.util.control.NonFatal
 
 import org.apache.spark._
 import org.apache.spark.broadcast.Broadcast
 import org.apache.spark.rdd.RDD
-import org.apache.spark.shuffle.ShuffleWriter
+import org.apache.spark.shuffle.{BinaryShuffleWriter, IndexShuffleBlockResolver, ShuffleWriter}
 
 /**
 * A ShuffleMapTask divides the elements of an RDD into multiple buckets (based on a partitioner
@@ -34,7 +35,7 @@ import org.apache.spark.shuffle.ShuffleWriter
 *
  * @param stageId id of the stage this task belongs to
  * @param taskBinary broadcast version of the RDD and the ShuffleDependency. Once deserialized,
- *                   the type should be (RDD[_], ShuffleDependency[_, _, _]).
+ *                   the type should be (RDD[_], BaseShuffleDependency[_]).
  * @param partition partition of the RDD this task is associated with
  * @param locs preferred task execution locations for locality scheduling
  */
@@ -61,28 +62,52 @@ private[spark] class ShuffleMapTask(
     // Deserialize the RDD using the broadcast variable.
     val deserializeStartTime = System.currentTimeMillis()
     val ser = SparkEnv.get.closureSerializer.newInstance()
-    val (rdd, dep) = ser.deserialize[(RDD[_], ShuffleDependency[_, _, _])](
+    val (rdd, dep) = ser.deserialize[(RDD[_], BaseShuffleDependency[_])](
       ByteBuffer.wrap(taskBinary.value), Thread.currentThread.getContextClassLoader)
     _executorDeserializeTime = System.currentTimeMillis() - deserializeStartTime
-
     metrics = Some(context.taskMetrics)
-    var writer: ShuffleWriter[Any, Any] = null
-    try {
-      val manager = SparkEnv.get.shuffleManager
-      writer = manager.getWriter[Any, Any](dep.shuffleHandle, partitionId, context)
-      writer.write(rdd.iterator(partition, context).asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
-      writer.stop(success = true).get
-    } catch {
-      case e: Exception =>
+
+    dep match {
+      case objDep: ShuffleDependency[_, _, _] =>
+        var writer: ShuffleWriter[Any, Any] = null
+        try {
+          val manager = SparkEnv.get.shuffleManager
+          writer = manager.getWriter[Any, Any](objDep.shuffleHandle, partitionId, context)
+          writer.write(rdd.iterator(partition, context)
+            .asInstanceOf[Iterator[_ <: Product2[Any, Any]]])
+          writer.stop(success = true).get
+        } catch {
+          case NonFatal(e) =>
+            try {
+              if (writer != null) {
+                writer.stop(success = false)
+              }
+            } catch {
+              case NonFatal(e2) =>
+                log.debug("Could not stop writer", e2)
+            }
+            throw e
+        }
+      case binaryDep: BinaryShuffleDependency[_] =>
+        // TODO(josh): Work around this limitation / document it.
+        val blockResolver = SparkEnv.get.shuffleManager.shuffleBlockResolver
+          .asInstanceOf[IndexShuffleBlockResolver]
+        val binaryWriter = new BinaryShuffleWriter(blockResolver, dep.shuffleId, partition.index)
+        val writeFunc = binaryDep.writeFunc.asInstanceOf[
+          (TaskContext, BinaryShuffleWriter, Iterator[Any]) => Unit]
         try {
-          if (writer != null) {
-            writer.stop(success = false)
-          }
+          writeFunc(context, binaryWriter, rdd.iterator(partition, context))
+          MapStatus(SparkEnv.get.blockManager.blockManagerId, binaryWriter.getPartitionLengths)
         } catch {
-          case e: Exception =>
-            log.debug("Could not stop writer", e)
+          case NonFatal(e) =>
+            try {
+              binaryWriter.abort()
+            } catch {
+            case NonFatal(e2) =>
+              log.debug("Could not abort / revert writes", e2)
+            }
+            throw e
         }
-        throw e
     }
   }
 
diff --git a/core/src/main/scala/org/apache/spark/shuffle/BinaryShuffleWriter.scala b/core/src/main/scala/org/apache/spark/shuffle/BinaryShuffleWriter.scala
new file mode 100644
index 0000000000000..ca70a7c3a77d1
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/shuffle/BinaryShuffleWriter.scala
@@ -0,0 +1,79 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle
+
+import java.io.File
+
+/**
+ * A low-level interface for writing shuffle output. This interface only exposes the path of the
+ * data file where data should be written, plus two methods for signalling the completion of a
+ * shuffle write or for aborting a failed write.
+ *
+ * The purpose of this low-level interface is to allow shuffle write to be performed by native
+ * (non-JVM) code or by external processes.
+ */
+private[spark] class BinaryShuffleWriter(
+    indexShuffleBlockResolver: IndexShuffleBlockResolver,
+    shuffleId: Int,
+    mapId: Int) {
+
+  private var aborted: Boolean = false
+  private var _partitionLengths: Array[Long] = _
+
+  /**
+   * Returns the file where shuffle data should be written. Within this file, the output partitions'
+   * data should be laid out contiguously in order of reduce partition id. In other words, the file
+   * should be laid out like [reduce partition 0's data][reduce partition 1's data]...
+   */
+  lazy val getDataFile: File = {
+    indexShuffleBlockResolver.getDataFile(shuffleId, mapId)
+  }
+
+  /**
+   * Call this method after writing a data file in order to signal a successful shuffle write.
+   *
+   * @param partitionLengths an array specifying the lengths of each reduce partition's data. These
+   *                         lengths will be used to serve regions of the shuffle output file to
+   *                         reducers.
+   */
+  def commit(partitionLengths: Array[Long]): Unit = {
+    require(_partitionLengths == null)
+    indexShuffleBlockResolver.writeIndexFile(shuffleId, mapId, partitionLengths)
+    _partitionLengths = partitionLengths
+  }
+
+  /**
+   * Call this method from error-handling code to perform cleanup after an unsuccessful shuffle
+   * write.
+   */
+  def abort(): Unit = {
+    require(_partitionLengths == null)
+    if (!aborted) {
+      aborted = true
+      indexShuffleBlockResolver.removeDataByMap(shuffleId, mapId)
+    }
+  }
+
+  /**
+   * Called by [[org.apache.spark.scheduler.ShuffleMapTask]] when constructing a MapStatus.
+   */
+  private[spark] def getPartitionLengths: Array[Long] = {
+    require(_partitionLengths != null)
+    _partitionLengths
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/shuffle/ObjectShuffleReader.scala b/core/src/main/scala/org/apache/spark/shuffle/ObjectShuffleReader.scala
new file mode 100644
index 0000000000000..6f1bc8fcdc964
--- /dev/null
+++ b/core/src/main/scala/org/apache/spark/shuffle/ObjectShuffleReader.scala
@@ -0,0 +1,98 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.shuffle
+
+import java.io.InputStream
+
+import org.apache.spark.util.CompletionIterator
+import org.apache.spark.util.collection.ExternalSorter
+import org.apache.spark.{InternalAccumulator, ShuffleDependency, InterruptibleIterator, TaskContext, SparkEnv}
+import org.apache.spark.io.CompressionCodec
+import org.apache.spark.serializer.Serializer
+
+
+private[spark] class ObjectShuffleReader[K, V, C](dep: ShuffleDependency[K, V, C]) {
+
+  private val conf = SparkEnv.get.conf
+
+  private val context = TaskContext.get()
+
+  def read(blockIter: Iterator[InputStream]): Iterator[Product2[K, C]] = {
+    // Wrap the streams for compression based on configuration
+    val wrappedStreams = blockIter.map { inputStream =>
+      if (conf.getBoolean("spark.shuffle.compress", true)) {
+        CompressionCodec.createCodec(conf).compressedInputStream(inputStream)
+      } else {
+        inputStream
+      }
+    }
+    val ser = Serializer.getSerializer(dep.serializer)
+    val serializerInstance = ser.newInstance()
+    // Create a key/value iterator for each stream
+    val recordIter = wrappedStreams.flatMap { wrappedStream =>
+      // Note: the asKeyValueIterator below wraps a key/value iterator inside of a
+      // NextIterator. The NextIterator makes sure that close() is called on the
+      // underlying InputStream when all records have been read.
+      serializerInstance.deserializeStream(wrappedStream).asKeyValueIterator
+    }
+    // Update the context task metrics for each record read.
+    val readMetrics = context.taskMetrics().createShuffleReadMetricsForDependency()
+    val metricIter = CompletionIterator[(Any, Any), Iterator[(Any, Any)]](
+      recordIter.map(record => {
+        readMetrics.incRecordsRead(1)
+        record
+      }),
+      context.taskMetrics().updateShuffleReadMetrics())
+
+    // An interruptible iterator must be used here in order to support task cancellation
+    val interruptibleIter = new InterruptibleIterator[(Any, Any)](context, metricIter)
+
+    val aggregatedIter: Iterator[Product2[K, C]] = if (dep.aggregator.isDefined) {
+      if (dep.mapSideCombine) {
+        // We are reading values that are already combined
+        val combinedKeyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, C)]]
+        dep.aggregator.get.combineCombinersByKey(combinedKeyValuesIterator, context)
+      } else {
+        // We don't know the value type, but also don't care -- the dependency *should*
+        // have made sure its compatible w/ this aggregator, which will convert the value
+        // type to the combined type C
+        val keyValuesIterator = interruptibleIter.asInstanceOf[Iterator[(K, Nothing)]]
+        dep.aggregator.get.combineValuesByKey(keyValuesIterator, context)
+      }
+    } else {
+      require(!dep.mapSideCombine, "Map-side combine without Aggregator specified!")
+      interruptibleIter.asInstanceOf[Iterator[Product2[K, C]]]
+    }
+
+    // Sort the output if there is a sort ordering defined.
+    dep.keyOrdering match {
+      case Some(keyOrd: Ordering[K]) =>
+        // Create an ExternalSorter to sort the data. Note that if spark.shuffle.spill is disabled,
+        // the ExternalSorter won't spill to disk.
+        val sorter = new ExternalSorter[K, C, C](ordering = Some(keyOrd), serializer = Some(ser))
+        sorter.insertAll(aggregatedIter)
+        context.taskMetrics().incMemoryBytesSpilled(sorter.memoryBytesSpilled)
+        context.taskMetrics().incDiskBytesSpilled(sorter.diskBytesSpilled)
+        context.internalMetricsToAccumulators(
+          InternalAccumulator.PEAK_EXECUTION_MEMORY).add(sorter.peakMemoryUsedBytes)
+        sorter.iterator
+      case None =>
+        aggregatedIter
+    }
+  }
+}
diff --git a/core/src/main/scala/org/apache/spark/shuffle/ShuffleReader.scala b/core/src/main/scala/org/apache/spark/shuffle/ShuffleReader.scala
index 292e48314ee10..114a9c9eb28e3 100644
--- a/core/src/main/scala/org/apache/spark/shuffle/ShuffleReader.scala
+++ b/core/src/main/scala/org/apache/spark/shuffle/ShuffleReader.scala
@@ -23,11 +23,4 @@ package org.apache.spark.shuffle
 private[spark] trait ShuffleReader[K, C] {
   /** Read the combined key-values for this reduce task */
   def read(): Iterator[Product2[K, C]]
-
-  /**
-   * Close this reader.
-   * TODO: Add this back when we make the ShuffleReader a developer API that others can implement
-   * (at which point this will likely be necessary).
-   */
-  // def stop(): Unit
 }
diff --git a/core/src/main/scala/org/apache/spark/shuffle/unsafe/UnsafeShuffleManager.scala b/core/src/main/scala/org/apache/spark/shuffle/unsafe/UnsafeShuffleManager.scala
index 75f22f642b9d1..23009675d3658 100644
--- a/core/src/main/scala/org/apache/spark/shuffle/unsafe/UnsafeShuffleManager.scala
+++ b/core/src/main/scala/org/apache/spark/shuffle/unsafe/UnsafeShuffleManager.scala
@@ -162,13 +162,17 @@ private[spark] class UnsafeShuffleManager(conf: SparkConf) extends ShuffleManage
       case unsafeShuffleHandle: UnsafeShuffleHandle[K @unchecked, V @unchecked] =>
         numMapsForShufflesThatUsedNewPath.putIfAbsent(handle.shuffleId, unsafeShuffleHandle.numMaps)
         val env = SparkEnv.get
+        val binaryWriter = new BinaryShuffleWriter(
+          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
+          handle.shuffleId,
+          mapId)
         new UnsafeShuffleWriter(
           env.blockManager,
-          shuffleBlockResolver.asInstanceOf[IndexShuffleBlockResolver],
+          binaryWriter,
           context.taskMemoryManager(),
           env.shuffleMemoryManager,
-          unsafeShuffleHandle,
-          mapId,
+          unsafeShuffleHandle.dependency.partitioner,
+          unsafeShuffleHandle.dependency.serializer,
           context,
           env.conf)
       case other =>
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
index 029f2264a6a27..f6432214aa3b0 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/Exchange.scala
@@ -130,7 +130,6 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
   @transient private lazy val sparkConf = child.sqlContext.sparkContext.getConf
 
   private val serializer: Serializer = {
-    val rowDataTypes = child.output.map(_.dataType).toArray
     if (tungstenMode) {
       new UnsafeRowSerializer(child.output.size)
     } else {
@@ -181,7 +180,17 @@ case class Exchange(newPartitioning: Partitioning, child: SparkPlan) extends Una
         }
       }
     }
-    new ShuffledRowRDD(rddWithPartitionIds, serializer, part.numPartitions)
+    val shuffleManager = SparkEnv.get.shuffleManager
+    val sortBasedShuffleOn = shuffleManager.isInstanceOf[SortShuffleManager] ||
+      shuffleManager.isInstanceOf[UnsafeShuffleManager]
+    if (sortBasedShuffleOn && tungstenMode) {
+      new UnsafeShuffledRowRDD(
+        rddWithPartitionIds.asInstanceOf[RDD[Product2[Int, UnsafeRow]]],
+        child.output.size,
+        part.numPartitions).asInstanceOf[RDD[InternalRow]]
+    } else {
+      new ShuffledRowRDD(rddWithPartitionIds, serializer, part.numPartitions)
+    }
   }
 }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
index 88f5b13c8f248..aa008d9b627d9 100644
--- a/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/ShuffledRowRDD.scala
@@ -32,7 +32,7 @@ private class ShuffledRowRDDPartition(val idx: Int) extends Partition {
  * A dummy partitioner for use with records whose partition ids have been pre-computed (i.e. for
  * use on RDDs of (Int, Row) pairs where the Int is a partition id in the expected range).
  */
-private class PartitionIdPassthrough(override val numPartitions: Int) extends Partitioner {
+private[execution] class PartitionIdPassthrough(override val numPartitions: Int) extends Partitioner {
   override def getPartition(key: Any): Int = key.asInstanceOf[Int]
 }
 
diff --git a/sql/core/src/main/scala/org/apache/spark/sql/execution/UnsafeShuffledRowRDD.scala b/sql/core/src/main/scala/org/apache/spark/sql/execution/UnsafeShuffledRowRDD.scala
new file mode 100644
index 0000000000000..e136d3be536e4
--- /dev/null
+++ b/sql/core/src/main/scala/org/apache/spark/sql/execution/UnsafeShuffledRowRDD.scala
@@ -0,0 +1,91 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.spark.sql.execution
+
+import java.io.InputStream
+
+import scala.util.control.NonFatal
+
+import org.apache.spark.shuffle.BinaryShuffleWriter
+import org.apache.spark.shuffle.unsafe.UnsafeShuffleWriter
+import org.apache.spark.sql.catalyst.expressions.UnsafeRow
+import org.apache.spark.{InterruptibleIterator, SparkEnv, TaskContext}
+import org.apache.spark.io.CompressionCodec
+import org.apache.spark.util.CompletionIterator
+import org.apache.spark.rdd.{BinaryShuffledRDD, RDD}
+
+
+class UnsafeShuffledRowRDD(
+    prev: RDD[Product2[Int, UnsafeRow]],
+    numFields: Int,
+    numPartitions: Int)
+  extends BinaryShuffledRDD[Product2[Int, UnsafeRow], UnsafeRow](prev, numPartitions) {
+
+  private val ser = new UnsafeRowSerializer(numFields)
+
+  override def write(
+      context: TaskContext,
+      binaryWriter: BinaryShuffleWriter,
+      iter: Iterator[Product2[Int, UnsafeRow]]): Unit = {
+    val env = SparkEnv.get
+    val writer = new UnsafeShuffleWriter[Int, UnsafeRow](
+      env.blockManager,
+      binaryWriter,
+      context.taskMemoryManager(),
+      env.shuffleMemoryManager,
+      new PartitionIdPassthrough(numPartitions),
+      Some(ser),
+      context,
+      env.conf)
+    try {
+      writer.write(iter)
+      writer.stop(true)
+    } catch {
+      case NonFatal(e) =>
+        writer.stop(false)
+    }
+  }
+
+  override def read(context: TaskContext, blockIter: Iterator[InputStream]): Iterator[UnsafeRow] = {
+    val wrappedStreams = blockIter.map { inputStream =>
+      // TODO: configure / respect compression settings
+      CompressionCodec.createCodec(SparkEnv.get.conf).compressedInputStream(inputStream)
+    }
+    val serializerInstance = ser.newInstance()
+
+    // Create a key/value iterator for each stream
+    val recordIter = wrappedStreams.flatMap { wrappedStream =>
+      // Note: the asKeyValueIterator below wraps a key/value iterator inside of a
+      // NextIterator. The NextIterator makes sure that close() is called on the
+      // underlying InputStream when all records have been read.
+      serializerInstance.deserializeStream(wrappedStream).asKeyValueIterator
+    }.map(_._2).asInstanceOf[Iterator[UnsafeRow]]
+
+    // Update the context task metrics for each record read.
+    val readMetrics = context.taskMetrics().createShuffleReadMetricsForDependency()
+    val metricIter = CompletionIterator[UnsafeRow, Iterator[UnsafeRow]](
+      recordIter.map(record => {
+        readMetrics.incRecordsRead(1)
+        record
+      }),
+      context.taskMetrics().updateShuffleReadMetrics())
+
+    // An interruptible iterator must be used here in order to support task cancellation
+    new InterruptibleIterator[UnsafeRow](context, metricIter)
+  }
+}