diff --git a/lang/cs/Org.Apache.REEF.IO.Tests/TestFilePartitionInputDataSet.cs b/lang/cs/Org.Apache.REEF.IO.Tests/TestFilePartitionInputDataSet.cs
index 001c3225ef..6c339cf389 100644
--- a/lang/cs/Org.Apache.REEF.IO.Tests/TestFilePartitionInputDataSet.cs
+++ b/lang/cs/Org.Apache.REEF.IO.Tests/TestFilePartitionInputDataSet.cs
@@ -19,6 +19,7 @@
 using System.Collections.Generic;
 using System.Globalization;
 using System.IO;
+using System.Linq;
 using Org.Apache.REEF.IO.PartitionedData;
 using Org.Apache.REEF.IO.PartitionedData.FileSystem;
 using Org.Apache.REEF.IO.PartitionedData.FileSystem.Parameters;
@@ -95,7 +96,7 @@ public void TestEvaluatorSideWithMultipleFilesOnePartition()
                     Assert.NotNull(partition);
                     Assert.NotNull(partition.Id);
                     int count = 0;
-                    var e = partition.GetPartitionHandle();
+                    var e = partition.GetPartitionHandle().First();
                     foreach (var v in e)
                     {
                         Logger.Log(Level.Info, string.Format(CultureInfo.CurrentCulture, "Data read {0}: ", v));
@@ -173,7 +174,7 @@ public void TestWithByteDeserializer()
                         .GetInstance<IInputPartition<IEnumerable<byte>>>();
                 using (partition as IDisposable)
                 {
-                    var e = partition.GetPartitionHandle();
+                    var e = partition.GetPartitionHandle().First();
                     foreach (var v in e)
                     {
                         Logger.Log(Level.Info, string.Format(CultureInfo.CurrentCulture, "Data read {0}: ", v));
@@ -211,7 +212,7 @@ public void TestWithRowDeserializer()
                         .GetInstance<IInputPartition<IEnumerable<Row>>>();
                 using (partition as IDisposable)
                 {
-                    IEnumerable<Row> e = partition.GetPartitionHandle();
+                    IEnumerable<Row> e = partition.GetPartitionHandle().First();
 
                     foreach (var row in e)
                     {
@@ -256,7 +257,7 @@ public void TestTempFileFolderWithRowDeserializer()
                         .GetInstance<IInputPartition<IEnumerable<Row>>>();
                 using (partition as IDisposable)
                 {
-                    IEnumerable<Row> e = partition.GetPartitionHandle();
+                    IEnumerable<Row> e = partition.GetPartitionHandle().First();
 
                     foreach (var row in e)
                     {
diff --git a/lang/cs/Org.Apache.REEF.IO.Tests/TestRandomInputDataSet.cs b/lang/cs/Org.Apache.REEF.IO.Tests/TestRandomInputDataSet.cs
index 452c75e147..1ed0da6d06 100644
--- a/lang/cs/Org.Apache.REEF.IO.Tests/TestRandomInputDataSet.cs
+++ b/lang/cs/Org.Apache.REEF.IO.Tests/TestRandomInputDataSet.cs
@@ -17,6 +17,7 @@
 
 using System.Collections.Generic;
 using System.IO;
+using System.Linq;
 using Org.Apache.REEF.IO.PartitionedData;
 using Org.Apache.REEF.IO.PartitionedData.Random;
 using Org.Apache.REEF.Tang.Implementations.Tang;
@@ -89,9 +90,10 @@ public void TestEvaluatorSide()
                 Assert.NotNull(partition);
                 Assert.NotNull(partition.Id);
 
-                using (var partitionStream = partition.GetPartitionHandle())
+                var partitionStreams = partition.GetPartitionHandle();
+                Assert.NotNull(partitionStreams);
+                using (var partitionStream = partitionStreams.Single())
                 {
-                    Assert.NotNull(partitionStream);
                     Assert.True(partitionStream.CanRead);
                     Assert.False(partitionStream.CanWrite);
                     Assert.Equal(ExpectedNumberOfBytesPerPartition, partitionStream.Length);
diff --git a/lang/cs/Org.Apache.REEF.IO/CacheLevel.cs b/lang/cs/Org.Apache.REEF.IO/CacheLevel.cs
new file mode 100644
index 0000000000..8c1eeb4387
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/CacheLevel.cs
@@ -0,0 +1,45 @@
+// 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.
+
+namespace Org.Apache.REEF.IO
+{
+    /// <summary>
+    /// The level at which data is cached.
+    /// </summary>
+    public enum CacheLevel
+    {
+        /// <summary>
+        /// The data is deserialized.
+        /// </summary>
+        InMemoryMaterialized = 0,
+
+        /// <summary>
+        /// The data is in memory, but as MemoryStream.
+        /// </summary>
+        InMemoryAsStream = 1,
+
+        /// <summary>
+        /// The data is in disk.
+        /// </summary>
+        Disk = 2,
+        
+        /// <summary>
+        /// The data is not local.
+        /// </summary>
+        NotLocal = 3,
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/DataCache.cs b/lang/cs/Org.Apache.REEF.IO/DataCache.cs
new file mode 100644
index 0000000000..92b56f7526
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/DataCache.cs
@@ -0,0 +1,276 @@
+// 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.
+
+using System;
+using System.Collections.Generic;
+using System.IO;
+using Org.Apache.REEF.Tang.Annotations;
+using Org.Apache.REEF.Tang.Exceptions;
+using Org.Apache.REEF.Utilities;
+using Org.Apache.REEF.Utilities.Attributes;
+
+namespace Org.Apache.REEF.IO
+{
+    /// <summary>
+    /// A class that manages the caching of data at different layers of storage medium.
+    /// </summary>
+    [Unstable("0.15", "API contract may change.")]
+    public sealed class DataCache<T>
+    {
+        private readonly IDataMover<T> _dataMover;
+
+        private Optional<DirectoryInfo> _diskDirectory = Optional<DirectoryInfo>.Empty();
+        private Optional<IReadOnlyCollection<MemoryStream>> _memStreams = Optional<IReadOnlyCollection<MemoryStream>>.Empty();
+        private Optional<IReadOnlyCollection<T>> _materialized = Optional<IReadOnlyCollection<T>>.Empty();
+
+        private CacheLevel _cacheLevel = CacheLevel.NotLocal;
+
+        [Inject]
+        private DataCache(IDataMover<T> dataMover)
+        {
+            _dataMover = dataMover;
+        }
+
+        /// <summary>
+        /// The current level at which the data is cached.
+        /// </summary>
+        public CacheLevel CacheLevel
+        {
+            get { return _cacheLevel; }
+        }
+
+        /// <summary>
+        /// The directory where cached data resides, if the data is cached on disk.
+        /// </summary>
+        public Optional<DirectoryInfo> DiskDirectory
+        {
+            get { return _diskDirectory; }
+        }
+
+        /// <summary>
+        /// The collection of memory streams where cached data resides, if the data is
+        /// cached in memory.
+        /// </summary>
+        public Optional<IReadOnlyCollection<MemoryStream>> MemoryStreams
+        {
+            get { return _memStreams; }
+        }
+
+        /// <summary>
+        /// The cached collection of "materialized" objects, if the data is cached
+        /// in memory and deserialized.
+        /// </summary>
+        public Optional<IReadOnlyCollection<T>> Materialized
+        {
+            get { return _materialized; }
+        }
+
+        /// <summary>
+        /// Cache the data to a <see cref="CacheLevel"/>.
+        /// If the data is already cached at a lower level, no action will be taken.
+        /// </summary>
+        /// <param name="cacheToLevel">Level to cache to.</param>
+        /// <returns>The cached level.</returns>
+        public CacheLevel Cache(CacheLevel cacheToLevel)
+        {
+            if (_cacheLevel <= cacheToLevel)
+            {
+                return _cacheLevel;
+            }
+
+            switch (cacheToLevel)
+            {
+                case CacheLevel.NotLocal:
+                    return _cacheLevel;
+                case CacheLevel.Disk:
+                    return CacheToDisk();
+                case CacheLevel.InMemoryAsStream:
+                    return CacheToMemory();
+                case CacheLevel.InMemoryMaterialized:
+                    return CacheToMaterialized();
+                default:
+                    throw new SystemException("Unexpected cache level " + cacheToLevel);
+            }
+        }
+
+        /// <summary>
+        /// Cache to disk with the injected <see cref="IDataMover{T}"/>.
+        /// </summary>
+        /// <returns>The cached level.</returns>
+        private CacheLevel CacheToDisk()
+        {
+            switch (_cacheLevel)
+            {
+                case CacheLevel.NotLocal:
+                    _diskDirectory = Optional<DirectoryInfo>.Of(_dataMover.RemoteToDisk());
+                    break;
+                default:
+                    throw new SystemException(
+                        "Unexpected cache level transition from " + _cacheLevel + " to " + CacheLevel.Disk);
+            }
+
+            _cacheLevel = CacheLevel.Disk;
+            return _cacheLevel;
+        }
+
+        /// <summary>
+        /// Cache to memory with the injected <see cref="IDataMover{T}"/>.
+        /// </summary>
+        /// <returns>The cached level.</returns>
+        public CacheLevel CacheToMemory()
+        {
+            switch (_cacheLevel)
+            {
+                case CacheLevel.Disk:
+                    CheckDisk();
+                    _memStreams = Optional<IReadOnlyCollection<MemoryStream>>.Of(
+                        new List<MemoryStream>(_dataMover.DiskToMemory(_diskDirectory.Value)));
+                    CleanUpDisk();
+                    break;
+                case CacheLevel.NotLocal:
+                    _memStreams = Optional<IReadOnlyCollection<MemoryStream>>.Of(
+                        new List<MemoryStream>(_dataMover.RemoteToMemory()));
+                    break;
+                default:
+                    throw new SystemException(
+                        "Unexpected cache level transition from " + _cacheLevel + " to " + CacheLevel.InMemoryAsStream);
+            }
+
+            _cacheLevel = CacheLevel.InMemoryAsStream;
+            return _cacheLevel;
+        }
+
+        /// <summary>
+        /// "Materializes" and deserializes the data, with an option to cache.
+        /// </summary>
+        /// <returns>The deserialized data.</returns>
+        public IEnumerable<T> Materialize(bool shouldCache)
+        {
+            if (shouldCache)
+            {
+                CacheToMaterialized();
+            }
+
+            switch (_cacheLevel)
+            {
+                case CacheLevel.InMemoryAsStream:
+                    CheckMemory();
+                    return _dataMover.MemoryToMaterialized(_memStreams.Value);
+                case CacheLevel.Disk:
+                    CheckDisk();
+                    return _dataMover.DiskToMaterialized(_diskDirectory.Value);
+                case CacheLevel.NotLocal:
+                    return _dataMover.RemoteToMaterialized();
+                case CacheLevel.InMemoryMaterialized:
+                    CheckMaterialized();
+                    return _materialized.Value;
+                default:
+                    throw new IllegalStateException("Illegal cache layer.");
+            }
+        }
+
+        /// <summary>
+        /// Deserialize and cache the data in memory from the level the data is currently at.
+        /// </summary>
+        private CacheLevel CacheToMaterialized()
+        {
+            switch (_cacheLevel)
+            {
+                case CacheLevel.InMemoryAsStream:
+                    CheckMemory();
+                    _materialized = Optional<IReadOnlyCollection<T>>.Of(
+                        new List<T>(_dataMover.MemoryToMaterialized(_memStreams.Value)));
+                    CleanUpMemory();
+                    break;
+                case CacheLevel.Disk:
+                    CheckDisk();
+                    _materialized = Optional<IReadOnlyCollection<T>>.Of(
+                        new List<T>(_dataMover.DiskToMaterialized(_diskDirectory.Value)));
+                    CleanUpDisk();
+                    break;
+                case CacheLevel.NotLocal:
+                    _materialized = Optional<IReadOnlyCollection<T>>.Of(
+                        new List<T>(_dataMover.RemoteToMaterialized()));
+                    break;
+                default:
+                    throw new SystemException(
+                        "Unexpected cache level transition from " + _cacheLevel + " to " + CacheLevel.InMemoryMaterialized);
+            }
+
+            _cacheLevel = CacheLevel.InMemoryMaterialized;
+            return _cacheLevel;
+        }
+
+        /// <summary>
+        /// Check that the data is "materialized."
+        /// </summary>
+        private void CheckMaterialized()
+        {
+            if (!_materialized.IsPresent())
+            {
+                throw new IllegalStateException("Collection is expected to be materialized in memory.");
+            }
+        }
+
+        /// <summary>
+        /// Check that the data is in memory as a <see cref="MemoryStream"/>.
+        /// </summary>
+        private void CheckMemory()
+        {
+            if (!_memStreams.IsPresent())
+            {
+                throw new IllegalStateException("Data is expected to be in memory.");
+            }
+        }
+
+        /// <summary>
+        /// Clean up the data stored as <see cref="MemoryStream"/>s.
+        /// </summary>
+        private void CleanUpMemory()
+        {
+            CheckMemory();
+
+            foreach (var memStream in _memStreams.Value)
+            {
+                memStream.Dispose();
+            }
+
+            _memStreams = Optional<IReadOnlyCollection<MemoryStream>>.Empty();
+        }
+
+        /// <summary>
+        /// Check that the data is stored on disk.
+        /// </summary>
+        private void CheckDisk()
+        {
+            if (!_diskDirectory.IsPresent())
+            {
+                throw new IllegalStateException("Disk directory is expected to be present.");
+            }
+        }
+
+        /// <summary>
+        /// Clean up the data stored on disk.
+        /// </summary>
+        private void CleanUpDisk()
+        {
+            CheckDisk();
+            _diskDirectory.Value.Delete(true);
+            _diskDirectory = Optional<DirectoryInfo>.Empty();
+        }
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/FileStreamDataMover.cs b/lang/cs/Org.Apache.REEF.IO/FileStreamDataMover.cs
new file mode 100644
index 0000000000..a56e644bc8
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/FileStreamDataMover.cs
@@ -0,0 +1,62 @@
+// 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.
+
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using Org.Apache.REEF.Utilities.Attributes;
+
+namespace Org.Apache.REEF.IO
+{
+    /// <summary>
+    /// An abstract class inherits <see cref="SerializerBasedDataMover{T}"/> and models 
+    /// each file as a <see cref="Stream"/> (and consequently each file an object 
+    /// of type T.
+    /// </summary>
+    [Unstable("0.15", "API contract may change.")]
+    public abstract class FileStreamDataMover<T> : SerializerBasedDataMover<T>
+    {
+        protected FileStreamDataMover(ISerializer<T> serializer) 
+            : base(serializer)
+        {
+        }
+
+        public override IEnumerable<MemoryStream> DiskToMemory(DirectoryInfo info)
+        {
+            foreach (var file in info.EnumerateFiles("*", SearchOption.AllDirectories))
+            {
+                var memStream = new MemoryStream();
+                using (var fileStream = new FileStream(file.FullName, FileMode.Open))
+                {
+                    fileStream.CopyTo(memStream);
+                }
+
+                yield return memStream;
+            }
+        }
+
+        public override IEnumerable<T> DiskToMaterialized(DirectoryInfo info)
+        {
+            var fileStreams = 
+                info
+                .EnumerateFiles("*", SearchOption.AllDirectories)
+                .Select(file => new FileStream(file.FullName, FileMode.Open));
+
+            return StreamToMaterialized(fileStreams);
+        }
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/IDataMover.cs b/lang/cs/Org.Apache.REEF.IO/IDataMover.cs
new file mode 100644
index 0000000000..cdd1264522
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/IDataMover.cs
@@ -0,0 +1,66 @@
+// 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.
+
+using System.Collections.Generic;
+using System.IO;
+using Org.Apache.REEF.Utilities.Attributes;
+
+namespace Org.Apache.REEF.IO
+{
+    /// <summary>
+    /// An interface that describes a contract for data movement between 
+    /// different types of storage medium.
+    /// </summary>
+    [Unstable("0.15", "API contract may change.")]
+    public interface IDataMover<T>
+    {
+        /// <summary>
+        /// Fetches the data from remote and stores to disk.
+        /// </summary>
+        DirectoryInfo RemoteToDisk();
+
+        /// <summary>
+        /// Fetches the data from remote and puts in memory without deserialization.
+        /// </summary>
+        IEnumerable<MemoryStream> RemoteToMemory();
+
+        /// <summary>
+        /// Fetches the data from remote and deserializes it.
+        /// </summary>
+        IEnumerable<T> RemoteToMaterialized();
+
+        /// <summary>
+        /// Fetches the data from a directory on disk and puts in memory 
+        /// without deserialization.
+        /// </summary>
+        /// <param name="info">The directory.</param>
+        IEnumerable<MemoryStream> DiskToMemory(DirectoryInfo info);
+
+        /// <summary>
+        /// Fetches the data from a directory on disk and deserializes it.
+        /// </summary>
+        /// <param name="info"></param>
+        IEnumerable<T> DiskToMaterialized(DirectoryInfo info);
+
+        /// <summary>
+        /// Deserializes a memory stream.
+        /// </summary>
+        /// <param name="streams">Memory streams to deserialize.</param>
+        /// <returns></returns>
+        IEnumerable<T> MemoryToMaterialized(IEnumerable<MemoryStream> streams);
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/ISerializer.cs b/lang/cs/Org.Apache.REEF.IO/ISerializer.cs
new file mode 100644
index 0000000000..8f6c8e6f5f
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/ISerializer.cs
@@ -0,0 +1,29 @@
+// 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.
+
+using System.Collections.Generic;
+using System.IO;
+
+namespace Org.Apache.REEF.IO
+{
+    public interface ISerializer<T>
+    {
+        void Serialize(T value, out Stream stream);
+
+        T Deserialize(Stream stream);
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/Org.Apache.REEF.IO.csproj b/lang/cs/Org.Apache.REEF.IO/Org.Apache.REEF.IO.csproj
index 2fda1205e0..5dac181cda 100644
--- a/lang/cs/Org.Apache.REEF.IO/Org.Apache.REEF.IO.csproj
+++ b/lang/cs/Org.Apache.REEF.IO/Org.Apache.REEF.IO.csproj
@@ -70,6 +70,8 @@ under the License.
     <Reference Include="System.Xml" />
   </ItemGroup>
   <ItemGroup>
+    <Compile Include="CacheLevel.cs" />
+    <Compile Include="FileStreamDataMover.cs" />
     <Compile Include="FileSystem\AzureBlob\AzureBlockBlobFileSystemConfiguration.cs" />
     <Compile Include="FileSystem\AzureBlob\AzureBlobType.cs" />
     <Compile Include="FileSystem\AzureBlob\AzureBlockBlobFileSystemConfigurationProvider.cs" />
@@ -98,6 +100,10 @@ under the License.
     <Compile Include="FileSystem\AzureBlob\ICloudBlobContainer.cs" />
     <Compile Include="FileSystem\Local\LocalFileSystem.cs" />
     <Compile Include="FileSystem\Local\LocalFileSystemConfiguration.cs" />
+    <Compile Include="DataCache.cs" />
+    <Compile Include="IDataMover.cs" />
+    <Compile Include="ISerializer.cs" />
+    <Compile Include="PartitionedData\DefaultInputPartition.cs" />
     <Compile Include="PartitionedData\FileSystem\FileSystemInputPartition.cs" />
     <Compile Include="PartitionedData\FileSystem\FileInputPartitionDescriptor.cs" />
     <Compile Include="PartitionedData\FileSystem\Parameters\CopyToLocal.cs" />
@@ -118,6 +124,7 @@ under the License.
     <Compile Include="PartitionedData\Random\RandomInputPartition.cs" />
     <Compile Include="PartitionedData\Random\RandomInputPartitionDescriptor.cs" />
     <Compile Include="Properties\AssemblyInfo.cs" />
+    <Compile Include="SerializerBasedDataMover.cs" />
     <Compile Include="TempFileCreation\ITempFileCreator.cs" />
     <Compile Include="TempFileCreation\TempFileFolder.cs" />
     <Compile Include="TempFileCreation\TempFileConfigurationModule.cs" />
diff --git a/lang/cs/Org.Apache.REEF.IO/PartitionedData/DefaultInputPartition.cs b/lang/cs/Org.Apache.REEF.IO/PartitionedData/DefaultInputPartition.cs
new file mode 100644
index 0000000000..41652496a9
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/PartitionedData/DefaultInputPartition.cs
@@ -0,0 +1,53 @@
+// 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.
+
+using System.Collections.Generic;
+using Org.Apache.REEF.IO.PartitionedData.Random.Parameters;
+using Org.Apache.REEF.Tang.Annotations;
+
+namespace Org.Apache.REEF.IO.PartitionedData
+{
+    public sealed class DefaultInputPartition<T> : IInputPartition<T>
+    {
+        private readonly DataCache<T> _dataCache;
+        private readonly string _partitionId;
+
+        [Inject]
+        private DefaultInputPartition(
+            [Parameter(typeof(PartitionId))] string partitionId,
+            DataCache<T> dataCache)
+        {
+            _partitionId = partitionId;
+            _dataCache = dataCache;
+        }
+
+        public string Id
+        {
+            get { return _partitionId; }
+        }
+
+        public CacheLevel Cache(CacheLevel level)
+        {
+            return _dataCache.Cache(level);
+        }
+
+        public IEnumerable<T> GetPartitionHandle()
+        {
+            return _dataCache.Materialize(false);
+        }
+    }
+}
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/PartitionedData/FileSystem/FileSystemInputPartition.cs b/lang/cs/Org.Apache.REEF.IO/PartitionedData/FileSystem/FileSystemInputPartition.cs
index 0d17467a7e..a8e6a0beb5 100644
--- a/lang/cs/Org.Apache.REEF.IO/PartitionedData/FileSystem/FileSystemInputPartition.cs
+++ b/lang/cs/Org.Apache.REEF.IO/PartitionedData/FileSystem/FileSystemInputPartition.cs
@@ -26,7 +26,6 @@
 using Org.Apache.REEF.Tang.Annotations;
 using Org.Apache.REEF.Utilities;
 using Org.Apache.REEF.Utilities.Attributes;
-using Org.Apache.REEF.Utilities.Diagnostics;
 using Org.Apache.REEF.Utilities.Logging;
 
 namespace Org.Apache.REEF.IO.PartitionedData.FileSystem
@@ -71,13 +70,22 @@ public string Id
         /// <summary>
         /// Caches from the remote File System to a local disk.
         /// </summary>
-        public void Cache()
+        public CacheLevel Cache(CacheLevel level)
         {
             lock (_lock)
             {
-                if (!_localFiles.IsPresent())
+                if (level <= CacheLevel.Disk)
                 {
-                    _localFiles = Optional<ISet<string>>.Of(Download());
+                    if (!_localFiles.IsPresent())
+                    {
+                        _localFiles = Optional<ISet<string>>.Of(Download());
+                    }
+
+                    return CacheLevel.Disk;
+                }
+                else
+                {
+                    return _localFiles.IsPresent() ? CacheLevel.Disk : CacheLevel.NotLocal;
                 }
             }
         }
@@ -123,7 +131,7 @@ private ISet<string> Download()
         /// provided by the Serializer
         /// </summary>
         /// <returns></returns>
-        public T GetPartitionHandle()
+        public IEnumerable<T> GetPartitionHandle()
         {
             lock (_lock)
             {
@@ -131,15 +139,15 @@ public T GetPartitionHandle()
                 {
                     if (!_localFiles.IsPresent())
                     {
-                        Cache();
+                        Cache(CacheLevel.Disk);
                     }
 
                     // For now, assume IFileDeSerializer is local.
-                    return _fileSerializer.Deserialize(_localFiles.Value);
+                    return new List<T> { _fileSerializer.Deserialize(_localFiles.Value) };
                 }
 
                 // For now, assume IFileDeSerializer is remote.
-                return _fileSerializer.Deserialize(_remoteFilePaths);
+                return new List<T> { _fileSerializer.Deserialize(_remoteFilePaths) };
             }
         }
 
diff --git a/lang/cs/Org.Apache.REEF.IO/PartitionedData/IInputPartition.cs b/lang/cs/Org.Apache.REEF.IO/PartitionedData/IInputPartition.cs
index 33116cd617..ecd07e19e1 100644
--- a/lang/cs/Org.Apache.REEF.IO/PartitionedData/IInputPartition.cs
+++ b/lang/cs/Org.Apache.REEF.IO/PartitionedData/IInputPartition.cs
@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+using System.Collections.Generic;
 using Org.Apache.REEF.Utilities.Attributes;
 
 namespace Org.Apache.REEF.IO.PartitionedData
@@ -24,7 +25,8 @@ namespace Org.Apache.REEF.IO.PartitionedData
     /// </summary>
     /// <typeparam name="T">Generic Type representing data pointer.
     /// For example, for data in local file it can be file pointer </typeparam>
-    public interface IInputPartition<T>
+    [Unstable("API contract may change.")]
+    public interface IInputPartition<T> 
     {
         /// <summary>
         /// The id of the partition.
@@ -32,15 +34,15 @@ public interface IInputPartition<T>
         string Id { get; }
 
         /// <summary>
-        /// Caches the data locally, cached location is based on the implementation.
+        /// Caches the data based on the method parameter. Returns the actual cached level.
         /// </summary>
-        [Unstable("0.14", "Contract may change.")]
-        void Cache();
+        [Unstable("0.15", "Contract may change.")]
+        CacheLevel Cache(CacheLevel level);
 
         /// <summary>
         /// Gives a pointer to the underlying partition.
         /// </summary>
         /// <returns>The pointer to the underlying partition</returns>
-        T GetPartitionHandle();
+        IEnumerable<T> GetPartitionHandle();
     }
 }
\ No newline at end of file
diff --git a/lang/cs/Org.Apache.REEF.IO/PartitionedData/Random/RandomInputPartition.cs b/lang/cs/Org.Apache.REEF.IO/PartitionedData/Random/RandomInputPartition.cs
index df55555293..f76de6cf1a 100644
--- a/lang/cs/Org.Apache.REEF.IO/PartitionedData/Random/RandomInputPartition.cs
+++ b/lang/cs/Org.Apache.REEF.IO/PartitionedData/Random/RandomInputPartition.cs
@@ -16,6 +16,7 @@
 // under the License.
 
 using System;
+using System.Collections.Generic;
 using System.Diagnostics;
 using System.IO;
 using Org.Apache.REEF.IO.PartitionedData.Random.Parameters;
@@ -52,13 +53,18 @@ public string Id
             get { return _id; }
         }
 
-        public void Cache()
+        public CacheLevel Cache(CacheLevel level)
         {
             lock (_lock)
             {
+                if (level > CacheLevel.InMemoryMaterialized)
+                {
+                    return CacheLevel.NotLocal;
+                }
+
                 if (_randomData.IsPresent())
                 {
-                    return;
+                    return CacheLevel.InMemoryMaterialized;
                 }
 
                 var random = new System.Random();
@@ -75,20 +81,22 @@ public void Cache()
                         generatedData[index] = randomDoubleAsBytes[j];
                     }
                 }
+
                 _randomData = Optional<byte[]>.Of(generatedData);
+                return CacheLevel.InMemoryMaterialized;
             }
         }
 
-        public Stream GetPartitionHandle()
+        public IEnumerable<Stream> GetPartitionHandle()
         {
             lock (_lock)
             {
                 if (!_randomData.IsPresent())
                 {
-                    Cache();
+                    Cache(CacheLevel.InMemoryMaterialized);
                 }
 
-                return new MemoryStream(_randomData.Value, false);
+                return new List<Stream> { new MemoryStream(_randomData.Value, false) };
             }
         }
     }
diff --git a/lang/cs/Org.Apache.REEF.IO/SerializerBasedDataMover.cs b/lang/cs/Org.Apache.REEF.IO/SerializerBasedDataMover.cs
new file mode 100644
index 0000000000..b5a2854c01
--- /dev/null
+++ b/lang/cs/Org.Apache.REEF.IO/SerializerBasedDataMover.cs
@@ -0,0 +1,69 @@
+// 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.
+
+using System.Collections.Generic;
+using System.IO;
+using System.Linq;
+using Org.Apache.REEF.Utilities.Attributes;
+
+namespace Org.Apache.REEF.IO
+{
+    /// <summary>
+    /// A <see cref="IDataMover{T}"/> that uses <see cref="ISerializer{T}"/> to deserialize
+    /// data from <see cref="MemoryStream"/>.
+    /// </summary>
+    [Unstable("0.15", "API contract may change.")]
+    public abstract class SerializerBasedDataMover<T> : IDataMover<T>
+    {
+        private readonly ISerializer<T> _serializer;
+
+        protected SerializerBasedDataMover(ISerializer<T> serializer)
+        {
+            _serializer = serializer;
+        }
+
+        protected ISerializer<T> Serializer
+        {
+            get { return _serializer; }
+        }
+
+        public abstract DirectoryInfo RemoteToDisk();
+
+        public abstract IEnumerable<MemoryStream> RemoteToMemory();
+
+        public abstract IEnumerable<T> RemoteToMaterialized();
+
+        public abstract IEnumerable<MemoryStream> DiskToMemory(DirectoryInfo info);
+
+        public abstract IEnumerable<T> DiskToMaterialized(DirectoryInfo info);
+
+        public IEnumerable<T> MemoryToMaterialized(IEnumerable<MemoryStream> streams)
+        {
+            return StreamToMaterialized(streams);
+        }
+
+        /// <summary>
+        /// Uses an <see cref="ISerializer{T}"/> to deserialize data from <see cref="IEnumerable{MemoryStream}"/>,
+        /// where each stream is an object of type T.
+        /// </summary>
+        protected IEnumerable<T> StreamToMaterialized(IEnumerable<Stream> streams)
+        {
+            var deserialized = streams.Select(stream => _serializer.Deserialize(stream)).ToList();
+            return deserialized.AsReadOnly();
+        }
+    }
+}
\ No newline at end of file