[spark] branch master updated (731a60c -> c397b06)
This is an automated email from the ASF dual-hosted git repository.
srowen pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git.
from 731a60c [SPARK-27823][CORE][DOC][FOLLOWUP] Update doc of config
`spark.driver.resourcesFile`
add c397b06 [SPARK-28045][ML][PYTHON] add missing RankingEvaluator
No new revisions were added by this update.
Summary of changes:
.../spark/ml/evaluation/RankingEvaluator.scala | 118 +
...atorSuite.scala => RankingEvaluatorSuite.scala} | 46
python/pyspark/ml/evaluation.py| 95 -
3 files changed, 235 insertions(+), 24 deletions(-)
create mode 100644
mllib/src/main/scala/org/apache/spark/ml/evaluation/RankingEvaluator.scala
copy
mllib/src/test/scala/org/apache/spark/ml/evaluation/{MultilabelClassificationEvaluatorSuite.scala
=> RankingEvaluatorSuite.scala} (61%)
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (c397b06 -> 83b96f6)
This is an automated email from the ASF dual-hosted git repository. srowen pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git. from c397b06 [SPARK-28045][ML][PYTHON] add missing RankingEvaluator add 83b96f6 [SPARK-28117][ML] LDA and BisectingKMeans cache the input dataset if necessary No new revisions were added by this update. Summary of changes: .../spark/ml/clustering/BisectingKMeans.scala | 10 + .../scala/org/apache/spark/ml/clustering/LDA.scala | 25 +- .../spark/mllib/clustering/LDAOptimizer.scala | 6 +- .../spark/ml/clustering/BisectingKMeansSuite.scala | 2 +- 4 files changed, 36 insertions(+), 7 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated (83b96f6 -> c83b3dd)
This is an automated email from the ASF dual-hosted git repository. srowen pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git. from 83b96f6 [SPARK-28117][ML] LDA and BisectingKMeans cache the input dataset if necessary add c83b3dd [SPARK-28154][ML][FOLLOWUP] GMM fix double caching No new revisions were added by this update. Summary of changes: .../org/apache/spark/ml/clustering/GaussianMixture.scala | 12 ++-- 1 file changed, 10 insertions(+), 2 deletions(-) - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch branch-2.4 updated: [SPARK-28154][ML][FOLLOWUP] GMM fix double caching
This is an automated email from the ASF dual-hosted git repository.
srowen pushed a commit to branch branch-2.4
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/branch-2.4 by this push:
new eb97f95 [SPARK-28154][ML][FOLLOWUP] GMM fix double caching
eb97f95 is described below
commit eb97f952f6c54af4f192c7bf85a09612000ff793
Author: zhengruifeng
AuthorDate: Tue Jun 25 06:50:34 2019 -0500
[SPARK-28154][ML][FOLLOWUP] GMM fix double caching
if the input dataset is alreadly cached, then we do not need to cache the
internal rdd (like kmeans)
existing test
Closes #24919 from zhengruifeng/gmm_fix_double_caching.
Authored-by: zhengruifeng
Signed-off-by: Sean Owen
(cherry picked from commit c83b3ddb56d4a32158676b042a7eae861689e141)
Signed-off-by: Sean Owen
---
.../org/apache/spark/ml/clustering/GaussianMixture.scala | 12 ++--
1 file changed, 10 insertions(+), 2 deletions(-)
diff --git
a/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
b/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
index 88abc16..6c0b49c 100644
--- a/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
+++ b/mllib/src/main/scala/org/apache/spark/ml/clustering/GaussianMixture.scala
@@ -36,6 +36,7 @@ import org.apache.spark.rdd.RDD
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.sql.functions.udf
import org.apache.spark.sql.types.{IntegerType, StructType}
+import org.apache.spark.storage.StorageLevel
/**
@@ -342,10 +343,15 @@ class GaussianMixture @Since("2.0.0") (
val sc = dataset.sparkSession.sparkContext
val numClusters = $(k)
+val handlePersistence = dataset.storageLevel == StorageLevel.NONE
val instances = dataset
.select(DatasetUtils.columnToVector(dataset, getFeaturesCol)).rdd.map {
case Row(features: Vector) => features
-}.cache()
+}
+
+if (handlePersistence) {
+ instances.persist(StorageLevel.MEMORY_AND_DISK)
+}
// Extract the number of features.
val numFeatures = instances.first().size
@@ -422,8 +428,10 @@ class GaussianMixture @Since("2.0.0") (
logLikelihood = sums.logLikelihood // this is the freshly computed
log-likelihood
iter += 1
}
+if (handlePersistence) {
+ instances.unpersist()
+}
-instances.unpersist(false)
val gaussianDists = gaussians.map { case (mean, covVec) =>
val cov = GaussianMixture.unpackUpperTriangularMatrix(numFeatures,
covVec.values)
new MultivariateGaussian(mean, cov)
-
To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org
For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated: [SPARK-26985][CORE] Fix "access only some column of the all of columns " for big endian architecture BUG
This is an automated email from the ASF dual-hosted git repository.
srowen pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 1a3858a [SPARK-26985][CORE] Fix "access only some column of the all
of columns " for big endian architecture BUG
1a3858a is described below
commit 1a3858a7694a189f434846b132b828e902273620
Author: ketank-new
AuthorDate: Tue Jun 25 08:24:10 2019 -0500
[SPARK-26985][CORE] Fix "access only some column of the all of columns "
for big endian architecture BUG
continuation to https://github.com/apache/spark/pull/24788
## What changes were proposed in this pull request?
Changes are related to BIG ENDIAN system
This changes are done to
identify s390x platform.
use byteorder to BIG_ENDIAN for big endian systems
changes for 2 are done in access functions putFloats() and putDouble()
## How was this patch tested?
Changes have been tested to build successfully on s390x as well x86
platform to make sure build is successful.
Closes #24861 from ketank-new/ketan_latest_v2.3.2.
Authored-by: ketank-new
Signed-off-by: Sean Owen
---
common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java | 2 +-
.../apache/spark/sql/execution/vectorized/OffHeapColumnVector.java| 4 ++--
.../org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java | 4 ++--
3 files changed, 5 insertions(+), 5 deletions(-)
diff --git a/common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java
b/common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java
index d98f08f..dc8d6e3 100644
--- a/common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java
+++ b/common/unsafe/src/main/java/org/apache/spark/unsafe/Platform.java
@@ -304,7 +304,7 @@ public final class Platform {
static {
boolean _unaligned;
String arch = System.getProperty("os.arch", "");
-if (arch.equals("ppc64le") || arch.equals("ppc64")) {
+if (arch.equals("ppc64le") || arch.equals("ppc64") ||
arch.equals("s390x")) {
// Since java.nio.Bits.unaligned() doesn't return true on ppc (See
JDK-8165231), but
// ppc64 and ppc64le support it
_unaligned = true;
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
index 5e0cf7d..3b919c7 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OffHeapColumnVector.java
@@ -417,7 +417,7 @@ public final class OffHeapColumnVector extends
WritableColumnVector {
Platform.copyMemory(src, Platform.BYTE_ARRAY_OFFSET + srcIndex,
null, data + rowId * 4L, count * 4L);
} else {
- ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.LITTLE_ENDIAN);
+ ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.BIG_ENDIAN);
long offset = data + 4L * rowId;
for (int i = 0; i < count; ++i, offset += 4) {
Platform.putFloat(null, offset, bb.getFloat(srcIndex + (4 * i)));
@@ -472,7 +472,7 @@ public final class OffHeapColumnVector extends
WritableColumnVector {
Platform.copyMemory(src, Platform.BYTE_ARRAY_OFFSET + srcIndex,
null, data + rowId * 8L, count * 8L);
} else {
- ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.LITTLE_ENDIAN);
+ ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.BIG_ENDIAN);
long offset = data + 8L * rowId;
for (int i = 0; i < count; ++i, offset += 8) {
Platform.putDouble(null, offset, bb.getDouble(srcIndex + (8 * i)));
diff --git
a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
index 577eab6..1bf3126 100644
---
a/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
+++
b/sql/core/src/main/java/org/apache/spark/sql/execution/vectorized/OnHeapColumnVector.java
@@ -396,7 +396,7 @@ public final class OnHeapColumnVector extends
WritableColumnVector {
Platform.copyMemory(src, Platform.BYTE_ARRAY_OFFSET + srcIndex,
floatData,
Platform.DOUBLE_ARRAY_OFFSET + rowId * 4L, count * 4L);
} else {
- ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.LITTLE_ENDIAN);
+ ByteBuffer bb = ByteBuffer.wrap(src).order(ByteOrder.BIG_ENDIAN);
for (int i = 0; i < count; ++i) {
floatData[i + rowId] = bb.getFloat(srcIndex + (4 * i));
}
@@ -445,7 +445,7 @@ public final class OnHeapColumnVector extends
WritableColumnVector {
Platform.copyMemory(src, Platform.BYTE_ARRAY_OFFSET + srcIndex,
doubleData,
Platform.DOUBLE_ARRAY_OFFSET + rowId * 8L,
[spark] branch master updated (1a3858a -> b71c130)
This is an automated email from the ASF dual-hosted git repository. vanzin pushed a change to branch master in repository https://gitbox.apache.org/repos/asf/spark.git. from 1a3858a [SPARK-26985][CORE] Fix "access only some column of the all of columns " for big endian architecture BUG add b71c130 [SPARK-27622][CORE] Avoiding the network when block manager fetches disk persisted RDD blocks from the same host No new revisions were added by this update. Summary of changes: .../spark/network/shuffle/ExecutorDiskUtils.java | 66 +++ .../shuffle/ExternalShuffleBlockResolver.java | 45 +--- .../shuffle/ExternalShuffleBlockResolverSuite.java | 2 +- .../network/shuffle/TestShuffleDataContext.java| 9 +- .../org/apache/spark/storage/BlockManager.scala| 121 .../apache/spark/storage/BlockManagerMaster.scala | 13 ++- .../spark/storage/BlockManagerMasterEndpoint.scala | 32 -- .../spark/storage/BlockManagerMessages.scala | 17 ++- .../apache/spark/storage/DiskBlockManager.scala| 5 +- .../spark/storage/BlockManagerInfoSuite.scala | 1 + .../apache/spark/storage/BlockManagerSuite.scala | 124 +++-- 11 files changed, 342 insertions(+), 93 deletions(-) create mode 100644 common/network-shuffle/src/main/java/org/apache/spark/network/shuffle/ExecutorDiskUtils.java - To unsubscribe, e-mail: commits-unsubscr...@spark.apache.org For additional commands, e-mail: commits-h...@spark.apache.org
[spark] branch master updated: [SPARK-27630][CORE] Properly handle task end events from completed stages
This is an automated email from the ASF dual-hosted git repository.
irashid pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new 38263f6 [SPARK-27630][CORE] Properly handle task end events from
completed stages
38263f6 is described below
commit 38263f6d153944b6f2f0248d9284861fc82532d6
Author: sychen
AuthorDate: Tue Jun 25 14:30:13 2019 -0500
[SPARK-27630][CORE] Properly handle task end events from completed stages
## What changes were proposed in this pull request?
Track tasks separately for each stage attempt (instead of tracking by
stage), and do NOT reset the numRunningTasks to 0 on StageCompleted.
In the case of stage retry, the `taskEnd` event from the zombie stage
sometimes makes the number of `totalRunningTasks` negative, which will causes
the job to get stuck.
Similar problem also exists with `stageIdToTaskIndices` &
`stageIdToSpeculativeTaskIndices`.
If it is a failed `taskEnd` event of the zombie stage, this will cause
`stageIdToTaskIndices` or `stageIdToSpeculativeTaskIndices` to remove the task
index of the active stage, and the number of `totalPendingTasks` will increase
unexpectedly.
## How was this patch tested?
unit test properly handle task end events from completed stages
Closes #24497 from cxzl25/fix_stuck_job_follow_up.
Authored-by: sychen
Signed-off-by: Imran Rashid
---
.../apache/spark/ExecutorAllocationManager.scala | 113 -
.../org/apache/spark/scheduler/DAGScheduler.scala | 2 +-
.../org/apache/spark/scheduler/SparkListener.scala | 5 +-
.../spark/ExecutorAllocationManagerSuite.scala | 33 +-
project/MimaExcludes.scala | 6 ++
5 files changed, 104 insertions(+), 55 deletions(-)
diff --git
a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
index bb95fea..bceb26c 100644
--- a/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
+++ b/core/src/main/scala/org/apache/spark/ExecutorAllocationManager.scala
@@ -491,6 +491,10 @@ private[spark] class ExecutorAllocationManager(
numExecutorsToAdd = 1
}
+ private case class StageAttempt(stageId: Int, stageAttemptId: Int) {
+override def toString: String = s"Stage $stageId (Attempt $stageAttemptId)"
+ }
+
/**
* A listener that notifies the given allocation manager of when to add and
remove executors.
*
@@ -499,29 +503,32 @@ private[spark] class ExecutorAllocationManager(
*/
private[spark] class ExecutorAllocationListener extends SparkListener {
-private val stageIdToNumTasks = new mutable.HashMap[Int, Int]
-// Number of running tasks per stage including speculative tasks.
+private val stageAttemptToNumTasks = new mutable.HashMap[StageAttempt, Int]
+// Number of running tasks per stageAttempt including speculative tasks.
// Should be 0 when no stages are active.
-private val stageIdToNumRunningTask = new mutable.HashMap[Int, Int]
-private val stageIdToTaskIndices = new mutable.HashMap[Int,
mutable.HashSet[Int]]
-// Number of speculative tasks to be scheduled in each stage
-private val stageIdToNumSpeculativeTasks = new mutable.HashMap[Int, Int]
-// The speculative tasks started in each stage
-private val stageIdToSpeculativeTaskIndices = new mutable.HashMap[Int,
mutable.HashSet[Int]]
-
-// stageId to tuple (the number of task with locality preferences, a map
where each pair is a
-// node and the number of tasks that would like to be scheduled on that
node) map,
-// maintain the executor placement hints for each stage Id used by
resource framework to better
-// place the executors.
-private val stageIdToExecutorPlacementHints = new mutable.HashMap[Int,
(Int, Map[String, Int])]
+private val stageAttemptToNumRunningTask = new
mutable.HashMap[StageAttempt, Int]
+private val stageAttemptToTaskIndices = new mutable.HashMap[StageAttempt,
mutable.HashSet[Int]]
+// Number of speculative tasks to be scheduled in each stageAttempt
+private val stageAttemptToNumSpeculativeTasks = new
mutable.HashMap[StageAttempt, Int]
+// The speculative tasks started in each stageAttempt
+private val stageAttemptToSpeculativeTaskIndices =
+ new mutable.HashMap[StageAttempt, mutable.HashSet[Int]]
+
+// stageAttempt to tuple (the number of task with locality preferences, a
map where each pair
+// is a node and the number of tasks that would like to be scheduled on
that node) map,
+// maintain the executor placement hints for each stageAttempt used by
resource framework
+// to better place the executors.
+private val stageAttemptToExecutorPlacementHints =
+ new mutable.HashMap[StageAttempt, (Int, Map[String, Int])]
override def onStageSubm
[spark] branch master updated: [SPARK-27676][SQL][SS] InMemoryFileIndex should respect spark.sql.files.ignoreMissingFiles
This is an automated email from the ASF dual-hosted git repository.
gurwls223 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/spark.git
The following commit(s) were added to refs/heads/master by this push:
new d83f84a [SPARK-27676][SQL][SS] InMemoryFileIndex should respect
spark.sql.files.ignoreMissingFiles
d83f84a is described below
commit d83f84a1229138b7935b4b18da80a96d3c5c3dde
Author: Josh Rosen
AuthorDate: Wed Jun 26 09:11:28 2019 +0900
[SPARK-27676][SQL][SS] InMemoryFileIndex should respect
spark.sql.files.ignoreMissingFiles
## What changes were proposed in this pull request?
Spark's `InMemoryFileIndex` contains two places where `FileNotFound`
exceptions are caught and logged as warnings (during [directory
listing](https://github.com/apache/spark/blob/bcd3b61c4be98565352491a108e6394670a0f413/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources/InMemoryFileIndex.scala#L274)
and [block location
lookup](https://github.com/apache/spark/blob/bcd3b61c4be98565352491a108e6394670a0f413/sql/core/src/main/scala/org/apache/spark/sql/execution/datasources
[...]
I think that this is a dangerous default behavior because it can mask bugs
caused by race conditions (e.g. overwriting a table while it's being read) or
S3 consistency issues (there's more discussion on this in the [JIRA
ticket](https://issues.apache.org/jira/browse/SPARK-27676)). Failing fast when
we detect missing files is not sufficient to make concurrent table reads/writes
or S3 listing safe (there are other classes of eventual consistency issues to
worry about), but I think it's [...]
There may be some cases where users _do_ want to ignore missing files, but
I think that should be an opt-in behavior via the existing
`spark.sql.files.ignoreMissingFiles` flag (the current behavior is itself
race-prone because a file might be be deleted between catalog listing and query
execution time, triggering FileNotFoundExceptions on executors (which are
handled in a way that _does_ respect `ignoreMissingFIles`)).
This PR updates `InMemoryFileIndex` to guard the
log-and-ignore-FileNotFoundException behind the existing
`spark.sql.files.ignoreMissingFiles` flag.
**Note**: this is a change of default behavior, so I think it needs to be
mentioned in release notes.
## How was this patch tested?
New unit tests to simulate file-deletion race conditions, tested with both
values of the `ignoreMissingFIles` flag.
Closes #24668 from JoshRosen/SPARK-27676.
Lead-authored-by: Josh Rosen
Co-authored-by: Josh Rosen
Signed-off-by: HyukjinKwon
---
docs/sql-migration-guide-upgrade.md| 2 +
.../spark/sql/execution/command/CommandUtils.scala | 2 +-
.../execution/datasources/InMemoryFileIndex.scala | 71 ++--
.../sql/execution/datasources/FileIndexSuite.scala | 126 -
4 files changed, 188 insertions(+), 13 deletions(-)
diff --git a/docs/sql-migration-guide-upgrade.md
b/docs/sql-migration-guide-upgrade.md
index b062a04..c920f06 100644
--- a/docs/sql-migration-guide-upgrade.md
+++ b/docs/sql-migration-guide-upgrade.md
@@ -145,6 +145,8 @@ license: |
- Since Spark 3.0, a higher-order function `exists` follows the three-valued
boolean logic, i.e., if the `predicate` returns any `null`s and no `true` is
obtained, then `exists` will return `null` instead of `false`. For example,
`exists(array(1, null, 3), x -> x % 2 == 0)` will be `null`. The previous
behaviour can be restored by setting
`spark.sql.legacy.arrayExistsFollowsThreeValuedLogic` to `false`.
+ - Since Spark 3.0, if files or subdirectories disappear during recursive
directory listing (i.e. they appear in an intermediate listing but then cannot
be read or listed during later phases of the recursive directory listing, due
to either concurrent file deletions or object store consistency issues) then
the listing will fail with an exception unless
`spark.sql.files.ignoreMissingFiles` is `true` (default `false`). In previous
versions, these missing files or subdirectories would be i [...]
+
## Upgrading from Spark SQL 2.4 to 2.4.1
- The value of `spark.executor.heartbeatInterval`, when specified without
units like "30" rather than "30s", was
diff --git
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
index cac2519..b644e6d 100644
---
a/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
+++
b/sql/core/src/main/scala/org/apache/spark/sql/execution/command/CommandUtils.scala
@@ -67,7 +67,7 @@ object CommandUtils extends Logging {
override def accept(path: Path): Boolean = isDataPath(path,
stagingDir)
}
val fileStatusSeq = InMemoryFileIndex.bulkListLeafFiles(
- paths, sessionState.newHadoopC
