Merge remote-tracking branch 'upstream/master' into arrow-toPandas-stream-SPARK-23030

Author: BryanCutler

common/network-common/src/main/java/org/apache/spark/network/buffer/FileSegmentManagedBuffer.java

@@ -77,16 +77,16 @@ public ByteBuffer nioByteBuffer() throws IOException {
         return channel.map(FileChannel.MapMode.READ_ONLY, offset, length);
       }
     } catch (IOException e) {
+      String errorMessage = "Error in reading " + this;
       try {
         if (channel != null) {
           long size = channel.size();
-          throw new IOException("Error in reading " + this + " (actual file length " + size + ")",
-            e);
+          errorMessage = "Error in reading " + this + " (actual file length " + size + ")";
         }
       } catch (IOException ignored) {
         // ignore
       }
-      throw new IOException("Error in opening " + this, e);
+      throw new IOException(errorMessage, e);
     } finally {
       JavaUtils.closeQuietly(channel);
     }
@@ -95,26 +95,24 @@ public ByteBuffer nioByteBuffer() throws IOException {
   @Override
   public InputStream createInputStream() throws IOException {
     FileInputStream is = null;
+    boolean shouldClose = true;
     try {
       is = new FileInputStream(file);
       ByteStreams.skipFully(is, offset);
-      return new LimitedInputStream(is, length);
+      InputStream r = new LimitedInputStream(is, length);
+      shouldClose = false;
+      return r;
     } catch (IOException e) {
-      try {
-        if (is != null) {
-          long size = file.length();
-          throw new IOException("Error in reading " + this + " (actual file length " + size + ")",
-              e);
-        }
-      } catch (IOException ignored) {
-        // ignore
-      } finally {
+      String errorMessage = "Error in reading " + this;
+      if (is != null) {
+        long size = file.length();
+        errorMessage = "Error in reading " + this + " (actual file length " + size + ")";
+      }
+      throw new IOException(errorMessage, e);
+    } finally {
+      if (shouldClose) {
         JavaUtils.closeQuietly(is);
       }
-      throw new IOException("Error in opening " + this, e);
-    } catch (RuntimeException e) {
-      JavaUtils.closeQuietly(is);
-      throw e;
     }
   }
 

common/network-common/src/main/java/org/apache/spark/network/server/TransportServer.java

@@ -70,11 +70,14 @@ public TransportServer(
     this.appRpcHandler = appRpcHandler;
     this.bootstraps = Lists.newArrayList(Preconditions.checkNotNull(bootstraps));
 
+    boolean shouldClose = true;
     try {
       init(hostToBind, portToBind);
-    } catch (RuntimeException e) {
-      JavaUtils.closeQuietly(this);
-      throw e;
+      shouldClose = false;
+    } finally {
+      if (shouldClose) {
+        JavaUtils.closeQuietly(this);
+      }
     }
   }
 

core/src/main/scala/org/apache/spark/api/python/PythonRDD.scala

@@ -49,9 +49,6 @@ private[spark] class PythonRDD(
     isFromBarrier: Boolean = false)
   extends RDD[Array[Byte]](parent) {
 
-  val bufferSize = conf.getInt("spark.buffer.size", 65536)
-  val reuseWorker = conf.getBoolean("spark.python.worker.reuse", true)
-
   override def getPartitions: Array[Partition] = firstParent.partitions
 
   override val partitioner: Option[Partitioner] = {
@@ -61,7 +58,7 @@ private[spark] class PythonRDD(
   val asJavaRDD: JavaRDD[Array[Byte]] = JavaRDD.fromRDD(this)
 
   override def compute(split: Partition, context: TaskContext): Iterator[Array[Byte]] = {
-    val runner = PythonRunner(func, bufferSize, reuseWorker)
+    val runner = PythonRunner(func)
     runner.compute(firstParent.iterator(split, context), split.index, context)
   }
 

core/src/main/scala/org/apache/spark/api/python/PythonRunner.scala

@@ -27,6 +27,7 @@ import scala.collection.JavaConverters._
 
 import org.apache.spark._
 import org.apache.spark.internal.Logging
+import org.apache.spark.internal.config.PYSPARK_EXECUTOR_MEMORY
 import org.apache.spark.security.SocketAuthHelper
 import org.apache.spark.util._
 
@@ -62,14 +63,20 @@ private[spark] object PythonEvalType {
  */
 private[spark] abstract class BasePythonRunner[IN, OUT](
     funcs: Seq[ChainedPythonFunctions],
-    bufferSize: Int,
-    reuseWorker: Boolean,
     evalType: Int,
     argOffsets: Array[Array[Int]])
   extends Logging {
 
   require(funcs.length == argOffsets.length, "argOffsets should have the same length as funcs")
 
+  private val conf = SparkEnv.get.conf
+  private val bufferSize = conf.getInt("spark.buffer.size", 65536)
+  private val reuseWorker = conf.getBoolean("spark.python.worker.reuse", true)
+  // each python worker gets an equal part of the allocation. the worker pool will grow to the
+  // number of concurrent tasks, which is determined by the number of cores in this executor.
+  private val memoryMb = conf.get(PYSPARK_EXECUTOR_MEMORY)
+      .map(_ / conf.getInt("spark.executor.cores", 1))
+
   // All the Python functions should have the same exec, version and envvars.
   protected val envVars = funcs.head.funcs.head.envVars
   protected val pythonExec = funcs.head.funcs.head.pythonExec
@@ -82,7 +89,7 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
   private[spark] var serverSocket: Option[ServerSocket] = None
 
   // Authentication helper used when serving method calls via socket from Python side.
-  private lazy val authHelper = new SocketAuthHelper(SparkEnv.get.conf)
+  private lazy val authHelper = new SocketAuthHelper(conf)
 
   def compute(
       inputIterator: Iterator[IN],
@@ -95,6 +102,9 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
     if (reuseWorker) {
       envVars.put("SPARK_REUSE_WORKER", "1")
     }
+    if (memoryMb.isDefined) {
+      envVars.put("PYSPARK_EXECUTOR_MEMORY_MB", memoryMb.get.toString)
+    }
     val worker: Socket = env.createPythonWorker(pythonExec, envVars.asScala.toMap)
     // Whether is the worker released into idle pool
     val released = new AtomicBoolean(false)
@@ -485,20 +495,17 @@ private[spark] abstract class BasePythonRunner[IN, OUT](
 
 private[spark] object PythonRunner {
 
-  def apply(func: PythonFunction, bufferSize: Int, reuseWorker: Boolean): PythonRunner = {
-    new PythonRunner(Seq(ChainedPythonFunctions(Seq(func))), bufferSize, reuseWorker)
+  def apply(func: PythonFunction): PythonRunner = {
+    new PythonRunner(Seq(ChainedPythonFunctions(Seq(func))))
   }
 }
 
 /**
  * A helper class to run Python mapPartition in Spark.
  */
-private[spark] class PythonRunner(
-    funcs: Seq[ChainedPythonFunctions],
-    bufferSize: Int,
-    reuseWorker: Boolean)
+private[spark] class PythonRunner(funcs: Seq[ChainedPythonFunctions])
   extends BasePythonRunner[Array[Byte], Array[Byte]](
-    funcs, bufferSize, reuseWorker, PythonEvalType.NON_UDF, Array(Array(0))) {
+    funcs, PythonEvalType.NON_UDF, Array(Array(0))) {
 
   protected override def newWriterThread(
       env: SparkEnv,

core/src/main/scala/org/apache/spark/internal/config/package.scala

@@ -114,6 +114,10 @@ package object config {
     .checkValue(_ >= 0, "The off-heap memory size must not be negative")
     .createWithDefault(0)
 
+  private[spark] val PYSPARK_EXECUTOR_MEMORY = ConfigBuilder("spark.executor.pyspark.memory")
+    .bytesConf(ByteUnit.MiB)
+    .createOptional
+
   private[spark] val IS_PYTHON_APP = ConfigBuilder("spark.yarn.isPython").internal()
     .booleanConf.createWithDefault(false)
 

core/src/main/scala/org/apache/spark/security/SocketAuthHelper.scala

@@ -42,43 +42,59 @@ private[spark] class SocketAuthHelper(conf: SparkConf) {
    * Read the auth secret from the socket and compare to the expected value. Write the reply back
    * to the socket.
    *
-   * If authentication fails, this method will close the socket.
+   * If authentication fails or error is thrown, this method will close the socket.
    *
    * @param s The client socket.
    * @throws IllegalArgumentException If authentication fails.
    */
   def authClient(s: Socket): Unit = {
-    // Set the socket timeout while checking the auth secret. Reset it before returning.
-    val currentTimeout = s.getSoTimeout()
+    var shouldClose = true
     try {
-      s.setSoTimeout(10000)
-      val clientSecret = readUtf8(s)
-      if (secret == clientSecret) {
-        writeUtf8("ok", s)
-      } else {
-        writeUtf8("err", s)
-        JavaUtils.closeQuietly(s)
+      // Set the socket timeout while checking the auth secret. Reset it before returning.
+      val currentTimeout = s.getSoTimeout()
+      try {
+        s.setSoTimeout(10000)
+        val clientSecret = readUtf8(s)
+        if (secret == clientSecret) {
+          writeUtf8("ok", s)
+          shouldClose = false
+        } else {
+          writeUtf8("err", s)
+          throw new IllegalArgumentException("Authentication failed.")
+        }
+      } finally {
+        s.setSoTimeout(currentTimeout)
       }
     } finally {
-      s.setSoTimeout(currentTimeout)
+      if (shouldClose) {
+        JavaUtils.closeQuietly(s)
+      }
     }
   }
 
   /**
    * Authenticate with a server by writing the auth secret and checking the server's reply.
    *
-   * If authentication fails, this method will close the socket.
+   * If authentication fails or error is thrown, this method will close the socket.
    *
    * @param s The socket connected to the server.
    * @throws IllegalArgumentException If authentication fails.
    */
   def authToServer(s: Socket): Unit = {
-    writeUtf8(secret, s)
+    var shouldClose = true
+    try {
+      writeUtf8(secret, s)
 
-    val reply = readUtf8(s)
-    if (reply != "ok") {
-      JavaUtils.closeQuietly(s)
-      throw new IllegalArgumentException("Authentication failed.")
+      val reply = readUtf8(s)
+      if (reply != "ok") {
+        throw new IllegalArgumentException("Authentication failed.")
+      } else {
+        shouldClose = false
+      }
+    } finally {
+      if (shouldClose) {
+        JavaUtils.closeQuietly(s)
+      }
     }
   }
 

core/src/test/scala/org/apache/spark/util/collection/OpenHashMapSuite.scala

@@ -194,4 +194,50 @@ class OpenHashMapSuite extends SparkFunSuite with Matchers {
     val numInvalidValues = map.iterator.count(_._2 == 0)
     assertResult(0)(numInvalidValues)
   }
+
+  test("distinguish between the 0/0.0/0L and null") {
+    val specializedMap1 = new OpenHashMap[String, Long]
+    specializedMap1("a") = null.asInstanceOf[Long]
+    specializedMap1("b") = 0L
+    assert(specializedMap1.contains("a"))
+    assert(!specializedMap1.contains("c"))
+    // null.asInstance[Long] will return 0L
+    assert(specializedMap1("a") === 0L)
+    assert(specializedMap1("b") === 0L)
+    // If the data type is in @specialized annotation, and
+    // the `key` is not be contained, the `map(key)` will return 0
+    assert(specializedMap1("c") === 0L)
+
+    val specializedMap2 = new OpenHashMap[String, Double]
+    specializedMap2("a") = null.asInstanceOf[Double]
+    specializedMap2("b") = 0.toDouble
+    assert(specializedMap2.contains("a"))
+    assert(!specializedMap2.contains("c"))
+    // null.asInstance[Double] will return 0.0
+    assert(specializedMap2("a") === 0.0)
+    assert(specializedMap2("b") === 0.0)
+    assert(specializedMap2("c") === 0.0)
+
+    val map1 = new OpenHashMap[String, Short]
+    map1("a") = null.asInstanceOf[Short]
+    map1("b") = 0.toShort
+    assert(map1.contains("a"))
+    assert(!map1.contains("c"))
+    // null.asInstance[Short] will return 0
+    assert(map1("a") === 0)
+    assert(map1("b") === 0)
+    // If the data type is not in @specialized annotation, and
+    // the `key` is not be contained, the `map(key)` will return null
+    assert(map1("c") === null)
+
+    val map2 = new OpenHashMap[String, Float]
+    map2("a") = null.asInstanceOf[Float]
+    map2("b") = 0.toFloat
+    assert(map2.contains("a"))
+    assert(!map2.contains("c"))
+    // null.asInstance[Float] will return 0.0
+    assert(map2("a") === 0.0)
+    assert(map2("b") === 0.0)
+    assert(map2("c") === null)
+  }
 }

docs/configuration.md

@@ -179,6 +179,18 @@ of the most common options to set are:
     (e.g. <code>2g</code>, <code>8g</code>).
   </td>
 </tr>
+<tr>
+ <td><code>spark.executor.pyspark.memory</code></td>
+  <td>Not set</td>
+  <td>
+    The amount of memory to be allocated to PySpark in each executor, in MiB
+    unless otherwise specified.  If set, PySpark memory for an executor will be
+    limited to this amount. If not set, Spark will not limit Python's memory use
+    and it is up to the application to avoid exceeding the overhead memory space
+    shared with other non-JVM processes. When PySpark is run in YARN, this memory
+    is added to executor resource requests.
+  </td>
+</tr>
 <tr>
  <td><code>spark.executor.memoryOverhead</code></td>
   <td>executorMemory * 0.10, with minimum of 384 </td>

docs/running-on-kubernetes.md

@@ -185,6 +185,36 @@ To use a secret through an environment variable use the following options to the
 --conf spark.kubernetes.executor.secretKeyRef.ENV_NAME=name:key
 ```
 
+## Using Kubernetes Volumes
+
+Starting with Spark 2.4.0, users can mount the following types of Kubernetes [volumes](https://kubernetes.io/docs/concepts/storage/volumes/) into the driver and executor pods:
+* [hostPath](https://kubernetes.io/docs/concepts/storage/volumes/#hostpath): mounts a file or directory from the host node’s filesystem into a pod.
+* [emptyDir](https://kubernetes.io/docs/concepts/storage/volumes/#emptydir): an initially empty volume created when a pod is assigned to a node.
+* [persistentVolumeClaim](https://kubernetes.io/docs/concepts/storage/volumes/#persistentvolumeclaim): used to mount a `PersistentVolume` into a pod.
+
+To mount a volume of any of the types above into the driver pod, use the following configuration property:
+
+```
+--conf spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].mount.path=<mount path>
+--conf spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].mount.readOnly=<true|false>
+``` 
+
+Specifically, `VolumeType` can be one of the following values: `hostPath`, `emptyDir`, and `persistentVolumeClaim`. `VolumeName` is the name you want to use for the volume under the `volumes` field in the pod specification.
+
+Each supported type of volumes may have some specific configuration options, which can be specified using configuration properties of the following form:
+
+```
+spark.kubernetes.driver.volumes.[VolumeType].[VolumeName].options.[OptionName]=<value>
+``` 
+
+For example, the claim name of a `persistentVolumeClaim` with volume name `checkpointpvc` can be specified using the following property:
+
+```
+spark.kubernetes.driver.volumes.persistentVolumeClaim.checkpointpvc.options.claimName=check-point-pvc-claim
+```
+
+The configuration properties for mounting volumes into the executor pods use prefix `spark.kubernetes.executor.` instead of `spark.kubernetes.driver.`. For a complete list of available options for each supported type of volumes, please refer to the [Spark Properties](#spark-properties) section below. 
+
 ## Introspection and Debugging
 
 These are the different ways in which you can investigate a running/completed Spark application, monitor progress, and
@@ -299,21 +329,15 @@ RBAC authorization and how to configure Kubernetes service accounts for pods, pl
 
 ## Future Work
 
-There are several Spark on Kubernetes features that are currently being incubated in a fork -
-[apache-spark-on-k8s/spark](https://github.com/apache-spark-on-k8s/spark), which are expected to eventually make it into
-future versions of the spark-kubernetes integration.
+There are several Spark on Kubernetes features that are currently being worked on or planned to be worked on. Those features are expected to eventually make it into future versions of the spark-kubernetes integration.
 
 Some of these include:
 
-* R
-* Dynamic Executor Scaling
+* Dynamic Resource Allocation and External Shuffle Service
 * Local File Dependency Management
 * Spark Application Management
 * Job Queues and Resource Management
 
-You can refer to the [documentation](https://apache-spark-on-k8s.github.io/userdocs/) if you want to try these features
-and provide feedback to the development team.
-
 # Configuration
 
 See the [configuration page](configuration.html) for information on Spark configurations.  The following configurations are

external/kafka-0-10-sql/src/main/scala/org/apache/spark/sql/kafka010/KafkaContinuousReadSupport.scala

@@ -227,7 +227,7 @@ class KafkaContinuousPartitionReader(
 
         // This is a failOnDataLoss exception. Retry if nextKafkaOffset is within the data range,
         // or if it's the endpoint of the data range (i.e. the "true" next offset).
-        case e: IllegalStateException  if e.getCause.isInstanceOf[OffsetOutOfRangeException] =>
+        case e: IllegalStateException if e.getCause.isInstanceOf[OffsetOutOfRangeException] =>
           val range = consumer.getAvailableOffsetRange()
           if (range.latest >= nextKafkaOffset && range.earliest <= nextKafkaOffset) {
             // retry