Skip to content

Commit 941ffcb

Browse files
yugan95Yu Gan
yugan95
and
Yu Gan
authored
apache#120 sql task skew detect (apache#121)
* apache#120 sql task skew detect * apache#120 code review * apache#120 update version r32 * apache#120 code review * apache#120 avoid npe Co-authored-by: Yu Gan <[email protected]>
1 parent 778bef4 commit 941ffcb

File tree

45 files changed

+225
-42
lines changed

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

45 files changed

+225
-42
lines changed

‎assembly/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
<parent>
2222
<groupId>org.apache.spark</groupId>
2323
<artifactId>spark-parent_2.11</artifactId>
24-
<version>2.4.1-kylin-r31</version>
24+
<version>2.4.1-kylin-r32</version>
2525
<relativePath>../pom.xml</relativePath>
2626
</parent>
2727

‎common/kvstore/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/network-common/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/network-shuffle/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/network-yarn/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/sketch/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/tags/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎common/unsafe/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -22,7 +22,7 @@
2222
<parent>
2323
<groupId>org.apache.spark</groupId>
2424
<artifactId>spark-parent_2.11</artifactId>
25-
<version>2.4.1-kylin-r31</version>
25+
<version>2.4.1-kylin-r32</version>
2626
<relativePath>../../pom.xml</relativePath>
2727
</parent>
2828

‎core/pom.xml‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -21,7 +21,7 @@
2121
<parent>
2222
<groupId>org.apache.spark</groupId>
2323
<artifactId>spark-parent_2.11</artifactId>
24-
<version>2.4.1-kylin-r31</version>
24+
<version>2.4.1-kylin-r32</version>
2525
<relativePath>../pom.xml</relativePath>
2626
</parent>
2727

‎core/src/main/scala/org/apache/spark/scheduler/TaskSchedulerImpl.scala‎

Lines changed: 13 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@ import org.apache.spark.internal.config
3333
import org.apache.spark.rpc.RpcEndpoint
3434
import org.apache.spark.scheduler.SchedulingMode.SchedulingMode
3535
import org.apache.spark.scheduler.TaskLocality.TaskLocality
36+
import org.apache.spark.status.config.TASK_SKEW_DETECT_ENABLED
3637
import org.apache.spark.storage.BlockManagerId
3738
import org.apache.spark.util.{AccumulatorV2, SystemClock, ThreadUtils, Utils}
3839

@@ -148,6 +149,8 @@ private[spark] class TaskSchedulerImpl(
148149

149150
private[scheduler] var barrierCoordinator: RpcEndpoint = null
150151

152+
private lazy val skewDetectEnabled = conf.get(TASK_SKEW_DETECT_ENABLED)
153+
151154
private def maybeInitBarrierCoordinator(): Unit = {
152155
if (barrierCoordinator == null) {
153156
barrierCoordinator = new BarrierCoordinator(barrierSyncTimeout, sc.listenerBus,
@@ -214,6 +217,16 @@ private[spark] class TaskSchedulerImpl(
214217
}
215218
schedulableBuilder.addTaskSetManager(manager, manager.taskSet.properties)
216219

220+
if(skewDetectEnabled) {
221+
val jobId = taskSet.priority
222+
val stageId = taskSet.stageId
223+
val stageAttemptId = taskSet.stageAttemptId
224+
val executionId = Option(taskSet.properties)
225+
.map(_.getProperty("spark.sql.execution.id"))
226+
logInfo(s"On tasks submitting stageId: $stageId, " +
227+
s"stageAttemptId: $stageAttemptId, executionId: $executionId, jobId: $jobId")
228+
}
229+
217230
if (!isLocal && !hasReceivedTask) {
218231
starvationTimer.scheduleAtFixedRate(new TimerTask() {
219232
override def run() {

0 commit comments

Comments
 (0)