Repository: spark
Updated Branches:
refs/heads/master 46b2550bc -> 3af894511
[SPARK-16759][CORE] Add a configuration property to pass caller contexts of
upstream applications into Spark
## What changes were proposed in this pull request?
Many applications take Spark as a computing engine and run on it. This PR adds
a configuration property `spark.log.callerContext` that can be used by Spark's
upstream applications (e.g. Oozie) to set up their caller contexts into Spark.
In the end, Spark will combine its own caller context with the caller contexts
of its upstream applications, and write them into Yarn RM log and HDFS audit
log.
The audit log has a config to truncate the caller contexts passed in (default
128). The caller contexts will be sent over rpc, so it should be concise. The
call context written into HDFS log and Yarn log consists of two parts: the
information `A` specified by Spark itself and the value `B` of
`spark.log.callerContext` property. Currently `A` typically takes 64 to 74
characters, so `B` can have up to 50 characters (mentioned in the doc
`running-on-yarn.md`)
## How was this patch tested?
Manual tests. I have run some Spark applications with `spark.log.callerContext`
configuration in Yarn client/cluster mode, and verified that the caller
contexts were written into Yarn RM log and HDFS audit log correctly.
The ways to configure `spark.log.callerContext` property:
- In spark-defaults.conf:
```
spark.log.callerContext infoSpecifiedByUpstreamApp
```
- In app's source code:
```
val spark = SparkSession
.builder
.appName("SparkKMeans")
.config("spark.log.callerContext", "infoSpecifiedByUpstreamApp")
.getOrCreate()
```
When running on Spark Yarn cluster mode, the driver is unable to pass
'spark.log.callerContext' to Yarn client and AM since Yarn client and AM have
already started before the driver performs `.config("spark.log.callerContext",
"infoSpecifiedByUpstreamApp")`.
The following example shows the command line used to submit a SparkKMeans
application and the corresponding records in Yarn RM log and HDFS audit log.
Command:
```
./bin/spark-submit --verbose --executor-cores 3 --num-executors 1 --master yarn
--deploy-mode client --class org.apache.spark.examples.SparkKMeans
examples/target/original-spark-examples_2.11-2.1.0-SNAPSHOT.jar
hdfs://localhost:9000/lr_big.txt 2 5
```
Yarn RM log:
<img width="1440" alt="screen shot 2016-10-19 at 9 12 03 pm"
src="https://cloud.githubusercontent.com/assets/8546874/19547050/7d2f278c-9649-11e6-9df8-8d5ff12609f0.png";>
HDFS audit log:
<img width="1400" alt="screen shot 2016-10-19 at 10 18 14 pm"
src="https://cloud.githubusercontent.com/assets/8546874/19547102/096060ae-964a-11e6-981a-cb28efd5a058.png";>
Author: Weiqing Yang <[email protected]>
Closes #15563 from weiqingy/SPARK-16759.
Project: http://git-wip-us.apache.org/repos/asf/spark/repo
Commit: http://git-wip-us.apache.org/repos/asf/spark/commit/3af89451
Tree: http://git-wip-us.apache.org/repos/asf/spark/tree/3af89451
Diff: http://git-wip-us.apache.org/repos/asf/spark/diff/3af89451
Branch: refs/heads/master
Commit: 3af894511be6fcc17731e28b284dba432fe911f5
Parents: 46b2550
Author: Weiqing Yang <[email protected]>
Authored: Fri Nov 11 18:36:23 2016 -0800
Committer: Mridul Muralidharan <[email protected]>
Committed: Fri Nov 11 18:36:23 2016 -0800
----------------------------------------------------------------------
.../apache/spark/internal/config/package.scala | 4 ++
.../scala/org/apache/spark/scheduler/Task.scala | 13 ++++-
.../scala/org/apache/spark/util/Utils.scala | 53 ++++++++++++--------
docs/configuration.md | 9 ++++
.../spark/deploy/yarn/ApplicationMaster.scala | 3 +-
.../org/apache/spark/deploy/yarn/Client.scala | 3 +-
6 files changed, 61 insertions(+), 24 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/core/src/main/scala/org/apache/spark/internal/config/package.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/internal/config/package.scala
b/core/src/main/scala/org/apache/spark/internal/config/package.scala
index 4a3e3d5..2951bdc 100644
--- a/core/src/main/scala/org/apache/spark/internal/config/package.scala
+++ b/core/src/main/scala/org/apache/spark/internal/config/package.scala
@@ -207,6 +207,10 @@ package object config {
.booleanConf
.createWithDefault(false)
+ private[spark] val APP_CALLER_CONTEXT =
ConfigBuilder("spark.log.callerContext")
+ .stringConf
+ .createOptional
+
private[spark] val FILES_MAX_PARTITION_BYTES =
ConfigBuilder("spark.files.maxPartitionBytes")
.doc("The maximum number of bytes to pack into a single partition when
reading files.")
.longConf
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/core/src/main/scala/org/apache/spark/scheduler/Task.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/scheduler/Task.scala
b/core/src/main/scala/org/apache/spark/scheduler/Task.scala
index 9385e3c..d39651a 100644
--- a/core/src/main/scala/org/apache/spark/scheduler/Task.scala
+++ b/core/src/main/scala/org/apache/spark/scheduler/Task.scala
@@ -26,6 +26,7 @@ import scala.collection.mutable.HashMap
import org.apache.spark._
import org.apache.spark.executor.TaskMetrics
+import org.apache.spark.internal.config.APP_CALLER_CONTEXT
import org.apache.spark.memory.{MemoryMode, TaskMemoryManager}
import org.apache.spark.metrics.MetricsSystem
import org.apache.spark.serializer.SerializerInstance
@@ -92,8 +93,16 @@ private[spark] abstract class Task[T](
kill(interruptThread = false)
}
- new CallerContext("TASK", appId, appAttemptId, jobId, Option(stageId),
Option(stageAttemptId),
- Option(taskAttemptId), Option(attemptNumber)).setCurrentContext()
+ new CallerContext(
+ "TASK",
+ SparkEnv.get.conf.get(APP_CALLER_CONTEXT),
+ appId,
+ appAttemptId,
+ jobId,
+ Option(stageId),
+ Option(stageAttemptId),
+ Option(taskAttemptId),
+ Option(attemptNumber)).setCurrentContext()
try {
runTask(context)
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/core/src/main/scala/org/apache/spark/util/Utils.scala
----------------------------------------------------------------------
diff --git a/core/src/main/scala/org/apache/spark/util/Utils.scala
b/core/src/main/scala/org/apache/spark/util/Utils.scala
index 1de66af..c27cbe3 100644
--- a/core/src/main/scala/org/apache/spark/util/Utils.scala
+++ b/core/src/main/scala/org/apache/spark/util/Utils.scala
@@ -2569,6 +2569,7 @@ private[util] object CallerContext extends Logging {
* @param from who sets up the caller context (TASK, CLIENT, APPMASTER)
*
* The parameters below are optional:
+ * @param upstreamCallerContext caller context the upstream application passes
in
* @param appId id of the app this task belongs to
* @param appAttemptId attempt id of the app this task belongs to
* @param jobId id of the job this task belongs to
@@ -2578,26 +2579,38 @@ private[util] object CallerContext extends Logging {
* @param taskAttemptNumber task attempt id
*/
private[spark] class CallerContext(
- from: String,
- appId: Option[String] = None,
- appAttemptId: Option[String] = None,
- jobId: Option[Int] = None,
- stageId: Option[Int] = None,
- stageAttemptId: Option[Int] = None,
- taskId: Option[Long] = None,
- taskAttemptNumber: Option[Int] = None) extends Logging {
-
- val appIdStr = if (appId.isDefined) s"_${appId.get}" else ""
- val appAttemptIdStr = if (appAttemptId.isDefined) s"_${appAttemptId.get}"
else ""
- val jobIdStr = if (jobId.isDefined) s"_JId_${jobId.get}" else ""
- val stageIdStr = if (stageId.isDefined) s"_SId_${stageId.get}" else ""
- val stageAttemptIdStr = if (stageAttemptId.isDefined)
s"_${stageAttemptId.get}" else ""
- val taskIdStr = if (taskId.isDefined) s"_TId_${taskId.get}" else ""
- val taskAttemptNumberStr =
- if (taskAttemptNumber.isDefined) s"_${taskAttemptNumber.get}" else ""
-
- val context = "SPARK_" + from + appIdStr + appAttemptIdStr +
- jobIdStr + stageIdStr + stageAttemptIdStr + taskIdStr +
taskAttemptNumberStr
+ from: String,
+ upstreamCallerContext: Option[String] = None,
+ appId: Option[String] = None,
+ appAttemptId: Option[String] = None,
+ jobId: Option[Int] = None,
+ stageId: Option[Int] = None,
+ stageAttemptId: Option[Int] = None,
+ taskId: Option[Long] = None,
+ taskAttemptNumber: Option[Int] = None) extends Logging {
+
+ private val context = prepareContext("SPARK_" +
+ from +
+ appId.map("_" + _).getOrElse("") +
+ appAttemptId.map("_" + _).getOrElse("") +
+ jobId.map("_JId_" + _).getOrElse("") +
+ stageId.map("_SId_" + _).getOrElse("") +
+ stageAttemptId.map("_" + _).getOrElse("") +
+ taskId.map("_TId_" + _).getOrElse("") +
+ taskAttemptNumber.map("_" + _).getOrElse("") +
+ upstreamCallerContext.map("_" + _).getOrElse(""))
+
+ private def prepareContext(context: String): String = {
+ // The default max size of Hadoop caller context is 128
+ lazy val len =
SparkHadoopUtil.get.conf.getInt("hadoop.caller.context.max.size", 128)
+ if (context == null || context.length <= len) {
+ context
+ } else {
+ val finalContext = context.substring(0, len)
+ logWarning(s"Truncated Spark caller context from $context to
$finalContext")
+ finalContext
+ }
+ }
/**
* Set up the caller context [[context]] by invoking Hadoop CallerContext
API of
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/docs/configuration.md
----------------------------------------------------------------------
diff --git a/docs/configuration.md b/docs/configuration.md
index 41c1778..ea99592 100644
--- a/docs/configuration.md
+++ b/docs/configuration.md
@@ -202,6 +202,15 @@ of the most common options to set are:
or remotely ("cluster") on one of the nodes inside the cluster.
</td>
</tr>
+<tr>
+ <td><code>spark.log.callerContext</code></td>
+ <td>(none)</td>
+ <td>
+ Application information that will be written into Yarn RM log/HDFS audit
log when running on Yarn/HDFS.
+ Its length depends on the Hadoop configuration
<code>hadoop.caller.context.max.size</code>. It should be concise,
+ and typically can have up to 50 characters.
+ </td>
+</tr>
</table>
Apart from these, the following properties are also available, and may be
useful in some situations:
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
----------------------------------------------------------------------
diff --git
a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
index f2b9dfb..918cc2d 100644
--- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
+++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/ApplicationMaster.scala
@@ -202,7 +202,8 @@ private[spark] class ApplicationMaster(
attemptID = Option(appAttemptId.getAttemptId.toString)
}
- new CallerContext("APPMASTER",
+ new CallerContext(
+ "APPMASTER", sparkConf.get(APP_CALLER_CONTEXT),
Option(appAttemptId.getApplicationId.toString),
attemptID).setCurrentContext()
logInfo("ApplicationAttemptId: " + appAttemptId)
http://git-wip-us.apache.org/repos/asf/spark/blob/3af89451/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
----------------------------------------------------------------------
diff --git a/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
b/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
index e77fa38..1b75688 100644
--- a/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
+++ b/yarn/src/main/scala/org/apache/spark/deploy/yarn/Client.scala
@@ -161,7 +161,8 @@ private[spark] class Client(
reportLauncherState(SparkAppHandle.State.SUBMITTED)
launcherBackend.setAppId(appId.toString)
- new CallerContext("CLIENT", Option(appId.toString)).setCurrentContext()
+ new CallerContext("CLIENT", sparkConf.get(APP_CALLER_CONTEXT),
+ Option(appId.toString)).setCurrentContext()
// Verify whether the cluster has enough resources for our AM
verifyClusterResources(newAppResponse)
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
![https://cloud.githubusercontent.com/assets/8546874/19547050/7d2f278c-9649-11e6-9df8-8d5ff12609f0.png"](https://cloud.githubusercontent.com/assets/8546874/19547050/7d2f278c-9649-11e6-9df8-8d5ff12609f0.png"){kind=link}
![https://cloud.githubusercontent.com/assets/8546874/19547102/096060ae-964a-11e6-981a-cb28efd5a058.png"](https://cloud.githubusercontent.com/assets/8546874/19547102/096060ae-964a-11e6-981a-cb28efd5a058.png"){kind=link}