add RunJob method to SparkContext; add SparkContext property to RDD (#590)

* add RunJob method to SparkContext; add SparkContext property to RDD

* test RDD.SparkContext

csharp/Adapter/Microsoft.Spark.CSharp/Core/RDD.cs

@@ -51,6 +51,17 @@ namespace Microsoft.Spark.CSharp.Core
             }
         }
 
+        /// <summary>
+        /// Return the SparkContext that created this RDD
+        /// </summary>
+        public SparkContext SparkContext
+        {
+            get
+            {
+                return sparkContext;
+            }
+        }
+
         /// <summary>
         /// Return whether this RDD has been cached or not
         /// </summary>

csharp/Adapter/Microsoft.Spark.CSharp/Core/SparkContext.cs

@@ -571,6 +571,17 @@ namespace Microsoft.Spark.CSharp.Core
             SparkContextProxy.SetLogLevel(logLevel);
         }
 
+        /// <summary>
+        /// Run a job on a given set of partitions of an RDD.
+        /// </summary>
+        /// <typeparam name="T"></typeparam>
+        /// <param name="rdd"></param>
+        /// <param name="partitions"></param>
+        public void RunJob<T>(RDD<T> rdd, IEnumerable<int> partitions)
+        {
+            SparkContextProxy.RunJob(rdd.RddProxy, partitions);
+        }
+
         /// <summary>
         /// Cancel active jobs for the specified group. See <see cref="SetJobGroup"/> for more information.
         /// </summary>

csharp/AdapterTest/SparkContextTest.cs

@@ -152,6 +152,23 @@ namespace AdapterTest
             Assert.IsNotNull(hadoopConf);
         }
 
+        [Test]
+        public void TestRunJob()
+        {
+            // Arrange
+            Mock<ISparkContextProxy> sparkContextProxy = new Mock<ISparkContextProxy>();
+            SparkContext sc = new SparkContext(sparkContextProxy.Object, null);
+            RDD<int> rdd = sc.Parallelize(new int[] {0, 1, 2, 3, 4, 5}, 2);
+            sparkContextProxy.Setup(m => m.RunJob(It.IsAny<IRDDProxy>(), It.IsAny<IEnumerable<int>>()));
+
+            // Act
+            int[] partitions = new int[] { 0, 1 };
+            rdd.SparkContext.RunJob(rdd, partitions);
+
+            // Assert
+            sparkContextProxy.Verify(m => m.RunJob(rdd.RddProxy, partitions), Times.Once);
+        }
+
         [Test]
         public void TestCancelAllJobs()
         {