You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@beam.apache.org by GitBox <gi...@apache.org> on 2021/01/13 11:52:25 UTC
[GitHub] [beam] tszerszen opened a new pull request #13743: [BEAM-11213] Beam metrics should be displayed in Spark UI
tszerszen opened a new pull request #13743:
URL: https://github.com/apache/beam/pull/13743
Create event log following Spark History Server format in default /tmp/spark-events log directory, so it could be read by Spark History Server.
**Please** add a meaningful description for your change here
------------------------
Thank you for your contribution! Follow this checklist to help us incorporate your contribution quickly and easily:
- [ ] [**Choose reviewer(s)**](https://beam.apache.org/contribute/#make-your-change) and mention them in a comment (`R: @username`).
- [ ] Format the pull request title like `[BEAM-XXX] Fixes bug in ApproximateQuantiles`, where you replace `BEAM-XXX` with the appropriate JIRA issue, if applicable. This will automatically link the pull request to the issue.
- [ ] Update `CHANGES.md` with noteworthy changes.
- [ ] If this contribution is large, please file an Apache [Individual Contributor License Agreement](https://www.apache.org/licenses/icla.pdf).
See the [Contributor Guide](https://beam.apache.org/contribute) for more tips on [how to make review process smoother](https://beam.apache.org/contribute/#make-reviewers-job-easier).
Post-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
Lang | SDK | Dataflow | Flink | Samza | Spark | Twister2
--- | --- | --- | --- | --- | --- | ---
Go | [](https://ci-beam.apache.org/job/beam_PostCommit_Go/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Go_VR_Spark/lastCompletedBuild/) | ---
Java | [](https://ci-beam.apache.org/job/beam_PostCommit_Java/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Dataflow_Java11/lastCompletedBuild/) | [](https://ci-beam
.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Flink_Java11/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Flink_Streaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Samza/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Spark/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_PVR_Spark_Batch/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_SparkStructuredStreaming/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Java_ValidatesRunner_Twister2/lastCompletedBuild/)
Python | [](https://ci-beam.apache.org/job/beam_PostCommit_Python36/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python37/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python38/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_VR_Dataflow_V2/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Py_ValCont/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_PVR_Flink_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_Python_VR_Spark/lastCompletedBuild/) | ---
XLang | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Direct/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Dataflow/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Flink/lastCompletedBuild/) | --- | [](https://ci-beam.apache.org/job/beam_PostCommit_XVR_Spark/lastCompletedBuild/) | ---
Pre-Commit Tests Status (on master branch)
------------------------------------------------------------------------------------------------
--- |Java | Python | Go | Website | Whitespace | Typescript
--- | --- | --- | --- | --- | --- | ---
Non-portable | [](https://ci-beam.apache.org/job/beam_PreCommit_Java_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Python_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonLint_Cron/lastCompletedBuild/)<br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocker_Cron/lastCompletedBuild/) <br>[](https://ci-beam.apache.org/job/beam_PreCommit_PythonDocs_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Go_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Website_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Whitespace_Cron/lastCompletedBuild/) | [](https://ci-beam.apache.org/job/beam_PreCommit_Typescript_Cron/lastCompletedBuild/)
Portable | --- | [](https://ci-beam.apache.org/job/beam_PreCommit_Portable_Python_Cron/lastCompletedBuild/) | --- | --- | --- | ---
See [.test-infra/jenkins/README](https://github.com/apache/beam/blob/master/.test-infra/jenkins/README.md) for trigger phrase, status and link of all Jenkins jobs.
GitHub Actions Tests Status (on master branch)
------------------------------------------------------------------------------------------------
[](https://github.com/apache/beam/actions?query=workflow%3A%22Build+python+source+distribution+and+wheels%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Python+Tests%22+branch%3Amaster+event%3Aschedule)
[](https://github.com/apache/beam/actions?query=workflow%3A%22Java+Tests%22+branch%3Amaster+event%3Aschedule)
See [CI.md](https://github.com/apache/beam/blob/master/CI.md) for more information about GitHub Actions CI.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r566050949
##########
File path: runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
##########
@@ -35,7 +35,8 @@ public void testRenderName() {
"myStep.one.two(three)", MetricName.named("myNameSpace//", "myName()")),
123,
456);
- String renderedName = new SparkBeamMetric().renderName(metricResult);
+ new SparkBeamMetric();
Review comment:
It's not necessary, sorry I've removed it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r568026000
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
Review comment:
I am referring to certain fields in the Spark API that you have filled in incorrectly. Here (in comments) are the values I think you should use instead:
```java
EventLoggingListener(
String appId /* jsc.getConf().getAppId() */,
java.net.URI logBaseDir,
SparkConf sparkConf,
org.apache.hadoop.conf.Configuration hadoopConf)
SparkListenerApplicationStart(
String appName /* pipelineOptions.as(ApplicationNameOptions.class).getAppName() */,
scala.Option<String> appId /* jsc.getConf().getAppId() */,
long time,
String sparkUser,
scala.Option<String> appAttemptId /* "1" */,
scala.Option<scala.collection.Map<String,String>> driverLogs)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r568825987
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +247,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jsc.getConf().getAppId(),
Review comment:
This should be appName, not appId. https://spark.apache.org/docs/2.4.7/api/java/org/apache/spark/scheduler/SparkListenerApplicationStart.html#SparkListenerApplicationStart-java.lang.String-scala.Option-long-java.lang.String-scala.Option-scala.Option-
```suggestion
pipelineOptions.as(ApplicationNameOptions.class).getAppName(),
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +134,34 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ long startTime = Instant.now().getMillis();
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jsc.getConf().getAppId(),
+ scala.Option.apply(jobInfo.jobName()),
Review comment:
appAttemptId is a Spark concept. jobInfo.jobName is a Beam concept, which is unrelated.
https://github.com/apache/spark/blob/d308794adb821d301847772de3ee1ef3166aaf5b/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala#L60
```suggestion
scala.Option.apply("1"),
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkJobServerDriver.java
##########
@@ -36,6 +36,7 @@
/** Spark runner-specific Configuration for the jobServer. */
public static class SparkServerConfiguration extends ServerConfiguration {
+
Review comment:
Nit: remove whitespace changes to this file.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib merged pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib merged pull request #13743:
URL: https://github.com/apache/beam/pull/13743
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567279548
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +134,33 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ scala.Option.apply(jobInfo.jobName()),
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkMasters = jsc.getConf().getAllWithPrefix("spark.master");
+ Tuple2<String, String>[] sparkExecutors = jsc.getConf().getAllWithPrefix("spark.executor.id");
+ for (int i = 0; i < sparkMasters.length; i++) {
+ eventLoggingListener.onExecutorAdded(
+ new SparkListenerExecutorAdded(
+ Instant.now().getMillis(),
+ sparkExecutors[i]._2(),
Review comment:
This assumes the number of spark.master is the same as the number of spark.executor.id. Which I don't think is a safe assumption.
In the Spark execution model, usually there is exactly one spark.master and possibly many executors. I'm not sure if it's ever possible for there to be multiple masters.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567454089
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkCommonPipelineOptions.java
##########
@@ -32,6 +32,8 @@
public interface SparkCommonPipelineOptions
extends PipelineOptions, StreamingOptions, ApplicationNameOptions {
String DEFAULT_MASTER_URL = "local[4]";
+ String DEFAULT_SPARK_HISTORY_DIR = "/tmp/spark-events/";
Review comment:
Remove `DEFAULT_SPARK_HISTORY_DIR` and `DEFAULT_EVENT_LOG_ENABLED` since they are unused now.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
+ scala.Option.apply(jobInfo.jobName()),
+ Instant.now().getMillis(),
Review comment:
This block runs after the job has completed, so this would report the start time being almost exactly the same as the end time.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
Review comment:
There are a number of (job or app) * (name or id) fields. It will be misleading to users if one of these fields is used in place of another. I think the following mapping is more accurate:
```java
appName = pipelineOptions.as(ApplicationNameOptions.class).getAppName();
appId = jsc.getConf().getAppId();
appAttemptId = "1";
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316348
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
Ok, the code will throw an exception if eventLogEnabled is true.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567464074
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
Review comment:
I don't get this part. What do you mean?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r564186788
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,18 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("Set it to true if event logs should be saved to Spark History Server directory")
+ @Default.String("false")
+ String getEventLogEnabled();
Review comment:
This should be a boolean, not a String.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
Review comment:
```suggestion
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
Review comment:
I missed that. Sounds good.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
You should be able to remove the `if (eventLoggingListener != null)` check altogether; instead it should be impossible for `eventLoggingListener` to ever be null. (Ideally we'd have the null checker verify that, but it is ignored on this class for some other reason.)
One solution is to create an EventLoggingListener regardless of the value of `getEventLogEnabled`, and just don't start it or do anything else with it if `getEventLogEnabled` is false.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
Review comment:
This logic is more complicated than it needs to be. `if (pipelineOptions.getEventLogEnabled())` should be checked before doing anything related to event logging.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
The only place you use `sparkExecutorID` is to log the `onExecutorAdded` event. So I imagine there should be one `onExecutorAdded` event per distinct `sparkExecutorID`.
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
It's fine to add the pipeline options for the Python and Go SDKs in separate PRs.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,9 +38,35 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
+ public static Map<String, String> renderAll(MetricResults metricResults) {
Review comment:
What's the difference between this and the existing `renderAll` method?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
You shouldn't need to inherit from `Option` or `Some`, can't you just instantiate it directly? https://stackoverflow.com/questions/5287451/how-to-call-scalas-option-constructors-from-java
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled().equals("true")) {
+ throw new RuntimeException("Failed to initialize Spark History Log Directory");
+ }
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
Review comment:
This exception should not be ignored; remove this `catch` block.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +301,97 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ startTime,
+ sparkUser,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new scala.Option<Map<String, String>>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> get() {
+ return JavaConverters.mapAsScalaMapConverter(
+ SparkBeamMetric.renderAll(result.metrics()))
+ .asScala();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ }));
+ eventLoggingListener.onApplicationEnd(
+ new SparkListenerApplicationEnd(Instant.now().getMillis()));
Review comment:
I'm pretty sure pipeline end time (and also start time for that matter) is itself a metric. To keep things consistent, it'd be better to use that metric here instead of `Instant.now()`.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -268,12 +447,30 @@ public static void main(String[] args) throws Exception {
}
private static class SparkPipelineRunnerConfiguration {
+ @Option(
+ name = "--event-log-enabled",
+ usage = "Set it to true if event logs should be saved to Spark History Server directory")
+ private String eventLogEnabled = "false";
+
+ String getEventLogEnabled() {
+ return this.eventLogEnabled;
+ }
+
@Option(
name = "--base-job-name",
usage =
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
Please remove the `--spark-history-dir` option from `SparkPipelineRunnerConfiguration` since it is unused and unneeded.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565461803
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,18 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("Set it to true if event logs should be saved to Spark History Server directory")
+ @Default.String("false")
+ String getEventLogEnabled();
Review comment:
I've changed it to boolean.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r566051297
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobInfo.jobName();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
Review comment:
I've change the logic to use `getAllWithPrefix("spark.executor.id")`.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobInfo.jobName();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
Review comment:
I've changed the logic to use `getAllWithPrefix("spark.executor.id")`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772741139
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r569605006
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +247,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jsc.getConf().getAppId(),
Review comment:
Done.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkJobServerDriver.java
##########
@@ -36,6 +36,7 @@
/** Spark runner-specific Configuration for the jobServer. */
public static class SparkServerConfiguration extends ServerConfiguration {
+
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316933
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
For now I don't know, do you have an idea? Or could you refer someone who could help?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r568306911
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
Review comment:
Thank you, it's done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r561332986
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -274,6 +472,15 @@ public static void main(String[] args) throws Exception {
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
This isn't used anywhere?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
Review comment:
We throw an exception if the directory is missing, instead of silently failing.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Should we set `spark.eventLog.dir` in the Spark conf? Or does that not matter? https://github.com/apache/beam/blob/d1c8c241d57293b1fe9baa39f3b8e807b3a45d3a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java#L88
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
If the user intends to use history logging, but we fail to set it up, we should always throw an exception to the user instead of silently failing and doing nothing.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
Review comment:
Why make this a URI first? Wouldn't it be easier to just do `new File(pipelineOptions.getSparkHistoryDir())`?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
What if there are multiple executors?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
There has to be a way to create an `Option` without all this boilerplate. I'm not too familiar with Scala, but I think `Some` might be better here? https://www.scala-lang.org/api/current/scala/Some.html
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
We may reuse the Spark context, in which case `jsc.startTime()` might not be an accurate measure of the application start time.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +299,118 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ HashMap<String, String> driverLogs = new HashMap<String, String>();
+ MetricResults metricResults = result.metrics();
+ for (MetricResult<DistributionResult> distributionResultMetricResult :
Review comment:
What about the other metric types (counters and gauges)?
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
I don't think we should add `sparkHistoryDir` as an argument to the job server.
(I don't think sparkMasterUrl is really necessary either, since it was already a pipeline option. But it's probably too late to remove it now.)
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Perhaps we should also add an option like `isEventLogEnabled` and make it `false` by default, like Spark's `spark.eventLog.enabled` property.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r569604902
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +134,34 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ long startTime = Instant.now().getMillis();
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jsc.getConf().getAppId(),
+ scala.Option.apply(jobInfo.jobName()),
Review comment:
Ok, that's changed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772603335
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r564186788
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,18 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("Set it to true if event logs should be saved to Spark History Server directory")
+ @Default.String("false")
+ String getEventLogEnabled();
Review comment:
This should be a boolean, not a String.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
Review comment:
```suggestion
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
Review comment:
I missed that. Sounds good.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
You should be able to remove the `if (eventLoggingListener != null)` check altogether; instead it should be impossible for `eventLoggingListener` to ever be null. (Ideally we'd have the null checker verify that, but it is ignored on this class for some other reason.)
One solution is to create an EventLoggingListener regardless of the value of `getEventLogEnabled`, and just don't start it or do anything else with it if `getEventLogEnabled` is false.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
Review comment:
This logic is more complicated than it needs to be. `if (pipelineOptions.getEventLogEnabled())` should be checked before doing anything related to event logging.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
The only place you use `sparkExecutorID` is to log the `onExecutorAdded` event. So I imagine there should be one `onExecutorAdded` event per distinct `sparkExecutorID`.
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
It's fine to add the pipeline options for the Python and Go SDKs in separate PRs.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,9 +38,35 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
+ public static Map<String, String> renderAll(MetricResults metricResults) {
Review comment:
What's the difference between this and the existing `renderAll` method?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
You shouldn't need to inherit from `Option` or `Some`, can't you just instantiate it directly? https://stackoverflow.com/questions/5287451/how-to-call-scalas-option-constructors-from-java
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled().equals("true")) {
+ throw new RuntimeException("Failed to initialize Spark History Log Directory");
+ }
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
Review comment:
This exception should not be ignored; remove this `catch` block.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +301,97 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ startTime,
+ sparkUser,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new scala.Option<Map<String, String>>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> get() {
+ return JavaConverters.mapAsScalaMapConverter(
+ SparkBeamMetric.renderAll(result.metrics()))
+ .asScala();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ }));
+ eventLoggingListener.onApplicationEnd(
+ new SparkListenerApplicationEnd(Instant.now().getMillis()));
Review comment:
I'm pretty sure pipeline end time (and also start time for that matter) is itself a metric. To keep things consistent, it'd be better to use that metric here instead of `Instant.now()`.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -268,12 +447,30 @@ public static void main(String[] args) throws Exception {
}
private static class SparkPipelineRunnerConfiguration {
+ @Option(
+ name = "--event-log-enabled",
+ usage = "Set it to true if event logs should be saved to Spark History Server directory")
+ private String eventLogEnabled = "false";
+
+ String getEventLogEnabled() {
+ return this.eventLogEnabled;
+ }
+
@Option(
name = "--base-job-name",
usage =
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
Please remove the `--spark-history-dir` option from `SparkPipelineRunnerConfiguration` since it is unused and unneeded.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r561332986
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -274,6 +472,15 @@ public static void main(String[] args) throws Exception {
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
This isn't used anywhere?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
Review comment:
We throw an exception if the directory is missing, instead of silently failing.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Should we set `spark.eventLog.dir` in the Spark conf? Or does that not matter? https://github.com/apache/beam/blob/d1c8c241d57293b1fe9baa39f3b8e807b3a45d3a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java#L88
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
If the user intends to use history logging, but we fail to set it up, we should always throw an exception to the user instead of silently failing and doing nothing.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
Review comment:
Why make this a URI first? Wouldn't it be easier to just do `new File(pipelineOptions.getSparkHistoryDir())`?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
What if there are multiple executors?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
There has to be a way to create an `Option` without all this boilerplate. I'm not too familiar with Scala, but I think `Some` might be better here? https://www.scala-lang.org/api/current/scala/Some.html
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
We may reuse the Spark context, in which case `jsc.startTime()` might not be an accurate measure of the application start time.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +299,118 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ HashMap<String, String> driverLogs = new HashMap<String, String>();
+ MetricResults metricResults = result.metrics();
+ for (MetricResult<DistributionResult> distributionResultMetricResult :
Review comment:
What about the other metric types (counters and gauges)?
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
I don't think we should add `sparkHistoryDir` as an argument to the job server.
(I don't think sparkMasterUrl is really necessary either, since it was already a pipeline option. But it's probably too late to remove it now.)
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Perhaps we should also add an option like `isEventLogEnabled` and make it `false` by default, like Spark's `spark.eventLog.enabled` property.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316740
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
It doesn't matter, however for consistency I think it would be good to configure it in such a way.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563321494
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
Pipeline options are coded across all SDKs, therefore two extra options require to change pipeline options in Python SDK, Java SDK, Go SDK. My proposition is to keep it that way, and change every SDKs pipeline options gradually in other smaller PRs. Could we do it in such a way or do you insist on doing all these changes here in this PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316167
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +299,118 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ HashMap<String, String> driverLogs = new HashMap<String, String>();
+ MetricResults metricResults = result.metrics();
+ for (MetricResult<DistributionResult> distributionResultMetricResult :
Review comment:
Thank you, they were add in the last commit.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Ok, I've done it.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
+ eventLoggingListener = null;
+ }
+ if (eventLoggingListener != null) {
Review comment:
Ok, the code will throw an exception if eventLogEnabled is true.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
I'll change it to Instant.now().getMillis()
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
I'll change it to `Instant.now().getMillis()`
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
It doesn't matter, however for consistency I think it would be good to configure it in such a way.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
Review comment:
For now I don't know, do you have an idea? Or could you refer someone who could help?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
I have an error `Cannot inherit from final scala.Some` when I use scala.Some instead of an scala.Option.
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
Pipeline options are coded across all SDKs, therefore two extra options require to change pipeline options in Python SDK, Java SDK, Go SDK. My proposition is to keep it that way, and change every SDKs pipeline options gradually in other smaller PRs. Could we do it in such a way or do you insist on doing all these changes here in this PR?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-766417268
R: @ibzib
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tysonjh commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tysonjh commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-759548175
R: @ibzib
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563283595
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -274,6 +472,15 @@ public static void main(String[] args) throws Exception {
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
@ibzib it only affects Spark History Server, another flaw is for now it only shows start and end time with accumulator value to download, native Spark History Server logs are not supported yet.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563285647
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
Review comment:
Because class EventLoggingListener takes eventLogDirectory which has to be the URI class object as it's 3rd argument.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565460458
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled().equals("true")) {
+ throw new RuntimeException("Failed to initialize Spark History Log Directory");
+ }
+ }
+ } catch (URISyntaxException e) {
+ e.printStackTrace();
Review comment:
I've removed it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r566051900
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,13 +38,11 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
- Map<String, ?> renderAll() {
+ public static Map<String, ?> renderAll(MetricResults metricResults) {
Review comment:
Ok, done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-768592646
I have not checked what is the current status on the classic runners, are the Beam metrics shown correctly? Do you think we can reuse this work (if they are not shown correctly) there too @tszerszen ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567464049
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
+ scala.Option.apply(jobInfo.jobName()),
+ Instant.now().getMillis(),
Review comment:
Ok, now I'm using startTime.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567409686
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +134,33 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ scala.Option.apply(jobInfo.jobName()),
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkMasters = jsc.getConf().getAllWithPrefix("spark.master");
+ Tuple2<String, String>[] sparkExecutors = jsc.getConf().getAllWithPrefix("spark.executor.id");
+ for (int i = 0; i < sparkMasters.length; i++) {
+ eventLoggingListener.onExecutorAdded(
+ new SparkListenerExecutorAdded(
+ Instant.now().getMillis(),
+ sparkExecutors[i]._2(),
Review comment:
@ibzib done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772832893
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565462086
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ eventLogDirectory,
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ } else {
+ eventLoggingListener = null;
Review comment:
The line has been removed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565461452
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,9 +38,35 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
+ public static Map<String, String> renderAll(MetricResults metricResults) {
Review comment:
We need `Map<String, String> ` in Spark History Server, I created a new method renderAllToString and reused renderAll logic in it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-766400885
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-770174523
R: @ibzib
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-766400885
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563317013
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists() && eventLogDirectoryFile.isDirectory()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobId,
+ new scala.Option<String>() {
Review comment:
I have an error `Cannot inherit from final scala.Some` when I use scala.Some instead of an scala.Option.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565463529
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +301,97 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ startTime,
+ sparkUser,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new scala.Option<Map<String, String>>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> get() {
+ return JavaConverters.mapAsScalaMapConverter(
+ SparkBeamMetric.renderAll(result.metrics()))
+ .asScala();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ }));
+ eventLoggingListener.onApplicationEnd(
+ new SparkListenerApplicationEnd(Instant.now().getMillis()));
Review comment:
When I printed out the results of renderAll method I didn't found such metrics for whole pipeline only for it's parts. Maybe not all metrics appear in renderAll method or I should filter for them specifically?
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +301,97 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ startTime,
+ sparkUser,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new scala.Option<Map<String, String>>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> get() {
+ return JavaConverters.mapAsScalaMapConverter(
+ SparkBeamMetric.renderAll(result.metrics()))
+ .asScala();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ }));
+ eventLoggingListener.onApplicationEnd(
+ new SparkListenerApplicationEnd(Instant.now().getMillis()));
Review comment:
When I printed out the results of renderAll method I didn't found such metrics for whole pipeline only for it's parts. Maybe not all metrics appear in renderAll method or should I filter for them specifically?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316609
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
I'll change it to Instant.now().getMillis()
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +140,79 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ Long startTime = jsc.startTime();
Review comment:
I'll change it to `Instant.now().getMillis()`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r566051673
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobInfo.jobName();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ if (!sparkExecutorID.equals(sparkConf._2())) {
Review comment:
Ok, I've changed this along with `getAllWithPrefix("spark.executor.id")`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r568825987
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +247,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jsc.getConf().getAppId(),
Review comment:
This should be appName, not appId. https://spark.apache.org/docs/2.4.7/api/java/org/apache/spark/scheduler/SparkListenerApplicationStart.html#SparkListenerApplicationStart-java.lang.String-scala.Option-long-java.lang.String-scala.Option-scala.Option-
```suggestion
pipelineOptions.as(ApplicationNameOptions.class).getAppName(),
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +134,34 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ long startTime = Instant.now().getMillis();
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jsc.getConf().getAppId(),
+ scala.Option.apply(jobInfo.jobName()),
Review comment:
appAttemptId is a Spark concept. jobInfo.jobName is a Beam concept, which is unrelated.
https://github.com/apache/spark/blob/d308794adb821d301847772de3ee1ef3166aaf5b/core/src/main/scala/org/apache/spark/scheduler/EventLoggingListener.scala#L60
```suggestion
scala.Option.apply("1"),
```
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkJobServerDriver.java
##########
@@ -36,6 +36,7 @@
/** Spark runner-specific Configuration for the jobServer. */
public static class SparkServerConfiguration extends ServerConfiguration {
+
Review comment:
Nit: remove whitespace changes to this file.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772863093
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772685732
R: @ibzib
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] iemejia commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
iemejia commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-769027755
Great to know @tszerszen, thanks . Sounds like a nice follow up issue to be created/fixed in case you feel motivated after this one is merged.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r567464004
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkCommonPipelineOptions.java
##########
@@ -32,6 +32,8 @@
public interface SparkCommonPipelineOptions
extends PipelineOptions, StreamingOptions, ApplicationNameOptions {
String DEFAULT_MASTER_URL = "local[4]";
+ String DEFAULT_SPARK_HISTORY_DIR = "/tmp/spark-events/";
Review comment:
Ok, I removed it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316167
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +299,118 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ HashMap<String, String> driverLogs = new HashMap<String, String>();
+ MetricResults metricResults = result.metrics();
+ for (MetricResult<DistributionResult> distributionResultMetricResult :
Review comment:
Thank you, they were add in the last commit.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565634941
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobInfo.jobName();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ if (!sparkExecutorID.equals(sparkConf._2())) {
Review comment:
The current logic is, "if the Spark executor id is not the same as the last executor id we saw, log it." It may be better to just log every Spark executor id. I couldn't find any documentation on `spark.executor.id`, but I assume there wouldn't be duplicates. And even if there are duplicates, I don't think logging them will harm anything.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,13 +38,11 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
- Map<String, ?> renderAll() {
+ public static Map<String, ?> renderAll(MetricResults metricResults) {
Review comment:
This method doesn't need to be public.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
Review comment:
This can all be replaced with `scala.Option.apply(jobInfo.jobName())` (and similar for other instances of scala.Option).
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobInfo.jobName();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new URI(pipelineOptions.getSparkHistoryDir()),
+ jsc.getConf(),
+ jsc.hadoopConfiguration());
+ eventLoggingListener.initializeLogIfNecessary(false, false);
+ eventLoggingListener.start();
+ scala.collection.immutable.Map<String, String> logUrlMap =
+ new scala.collection.immutable.HashMap<String, String>();
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
Review comment:
Nit: instead of `getAll`, can you use `getAllWithPrefix("spark.master")` and `getAllWithPrefix("spark.executor.id")`?
##########
File path: runners/spark/job-server/build.gradle
##########
@@ -73,6 +73,8 @@ runShadow {
args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
if (project.hasProperty('sparkMasterUrl'))
args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+ if (project.hasProperty('sparkHistoryDir'))
Review comment:
I still don't think `--spark-history-dir` or `--event-log-enabled` should be arguments to the job server. Duplicating pipeline options in the job server creates complexity without adding benefit.
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -213,6 +301,97 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
result);
metricsPusher.start();
+ if (eventLoggingListener != null) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobId,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ startTime,
+ sparkUser,
+ new scala.Option<String>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public String get() {
+ return jobName;
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ },
+ new scala.Option<Map<String, String>>() {
+ @Override
+ public boolean isEmpty() {
+ return false;
+ }
+
+ @Override
+ public Map<String, String> get() {
+ return JavaConverters.mapAsScalaMapConverter(
+ SparkBeamMetric.renderAll(result.metrics()))
+ .asScala();
+ }
+
+ @Override
+ public Object productElement(int i) {
+ return null;
+ }
+
+ @Override
+ public int productArity() {
+ return 0;
+ }
+
+ @Override
+ public boolean canEqual(Object o) {
+ return false;
+ }
+ }));
+ eventLoggingListener.onApplicationEnd(
+ new SparkListenerApplicationEnd(Instant.now().getMillis()));
Review comment:
I'm not sure, it's not a blocker for this PR though. Thanks for checking.
##########
File path: runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java
##########
@@ -35,7 +35,8 @@ public void testRenderName() {
"myStep.one.two(three)", MetricName.named("myNameSpace//", "myName()")),
123,
456);
- String renderedName = new SparkBeamMetric().renderName(metricResult);
+ new SparkBeamMetric();
Review comment:
Why is instantiating an unused `SparkBeamMetric` necessary?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r568306911
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -212,6 +246,22 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
pipelineOptions.as(MetricsOptions.class),
result);
metricsPusher.start();
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener.onApplicationStart(
+ new SparkListenerApplicationStart(
+ jobInfo.jobId(),
Review comment:
Thank you, it's done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r566053073
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +135,66 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener = null;
+ if (pipelineOptions.getEventLogEnabled()) {
+ eventLoggingListener =
+ new EventLoggingListener(
+ jobInfo.jobId(),
+ new scala.Option<String>() {
Review comment:
I'm not sure, I'll write about it on Slack.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-772550491
Run Java PreCommit
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] ibzib commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
ibzib commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565645231
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/metrics/SparkBeamMetric.java
##########
@@ -38,9 +38,35 @@
/**
* An adapter between the {@link MetricsContainerStepMap} and Codahale's {@link Metric} interface.
*/
-class SparkBeamMetric implements Metric {
+public class SparkBeamMetric implements Metric {
private static final String ILLEGAL_CHARACTERS = "[^A-Za-z0-9-]";
+ public static Map<String, String> renderAll(MetricResults metricResults) {
Review comment:
Nice 👍
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565461562
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -268,12 +447,30 @@ public static void main(String[] args) throws Exception {
}
private static class SparkPipelineRunnerConfiguration {
+ @Option(
+ name = "--event-log-enabled",
+ usage = "Set it to true if event logs should be saved to Spark History Server directory")
+ private String eventLogEnabled = "false";
+
+ String getEventLogEnabled() {
+ return this.eventLogEnabled;
+ }
+
@Option(
name = "--base-job-name",
usage =
"The job to run. This must correspond to a subdirectory of the jar's BEAM-PIPELINE "
+ "directory. *Only needs to be specified if the jar contains multiple pipelines.*")
private String baseJobName = null;
+
+ @Option(
+ name = "--spark-history-dir",
+ usage = "Spark history dir to store logs (e.g. /tmp/spark-events/)")
+ private String sparkHistoryDir = "/tmp/spark-events/";
Review comment:
Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r563316221
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java
##########
@@ -34,6 +34,12 @@
*/
public interface SparkPipelineOptions extends SparkCommonPipelineOptions {
+ @Description("The directory to save Spark History Server logs")
+ @Default.String("/tmp/spark-events/")
Review comment:
Ok, I've done it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-768417387
R: @ibzib
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on a change in pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on a change in pull request #13743:
URL: https://github.com/apache/beam/pull/13743#discussion_r565464269
##########
File path: runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java
##########
@@ -123,10 +137,84 @@ public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
"Will stage {} files. (Enable logging at DEBUG level to see which files will be staged.)",
pipelineOptions.getFilesToStage().size());
LOG.debug("Staging files: {}", pipelineOptions.getFilesToStage());
-
PortablePipelineResult result;
final JavaSparkContext jsc = SparkContextFactory.getSparkContext(pipelineOptions);
+ EventLoggingListener eventLoggingListener;
+ String jobId = jobInfo.jobId();
+ String jobName = jobInfo.jobName();
+ long startTime = Instant.now().getMillis();
+ String sparkUser = jsc.sparkUser();
+ String sparkMaster = "";
+ String sparkExecutorID = "";
+ Tuple2<String, String>[] sparkConfList = jsc.getConf().getAll();
+ for (Tuple2<String, String> sparkConf : sparkConfList) {
+ if (sparkConf._1().equals("spark.master")) {
+ sparkMaster = sparkConf._2();
+ } else if (sparkConf._1().equals("spark.executor.id")) {
+ sparkExecutorID = sparkConf._2();
+ }
+ }
+ try {
+ URI eventLogDirectory = new URI(pipelineOptions.getSparkHistoryDir());
+ File eventLogDirectoryFile = new File(eventLogDirectory.getPath());
+ if (eventLogDirectoryFile.exists()
+ && eventLogDirectoryFile.isDirectory()
+ && pipelineOptions.getEventLogEnabled().equals("true")) {
Review comment:
I've changed it to boolean and started checking with `if (pipelineOptions.getEventLogEnabled())` as you suggested.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [beam] tszerszen commented on pull request #13743: [BEAM-11213] Display Beam Metrics in Spark History Server
Posted by GitBox <gi...@apache.org>.
tszerszen commented on pull request #13743:
URL: https://github.com/apache/beam/pull/13743#issuecomment-769018822
@iemejia I think this work can be reused there, since it's calling native Spark EventLoggingListener.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org