[BEAM-11213] Display Beam Metrics in Spark History Server

runners/spark/job-server/build.gradle

@@ -73,6 +73,10 @@ runShadow {
     args += ["--clean-artifacts-per-job=${project.property('cleanArtifactsPerJob')}"]
   if (project.hasProperty('sparkMasterUrl'))
     args += ["--spark-master-url=${project.property('sparkMasterUrl')}"]
+  if (project.hasProperty('sparkHistoryDir'))
+    args += ["--spark-history-dir=${project.property('sparkHistoryDir')}"]
+  if (project.hasProperty('eventLogEnabled'))
+    args += ["--event-log-enabled=${project.property('eventLogEnabled')}"]
 
   systemProperties System.properties
 

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/";
+  boolean DEFAULT_EVENT_LOG_ENABLED = false;
 
   @Description("The url of the spark master to connect to, (e.g. spark://host:port, local[4]).")
   @Default.String(DEFAULT_MASTER_URL)

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkJobInvoker.java

@@ -70,6 +70,12 @@ protected JobInvocation invokeWithExecutor(
     if (sparkOptions.getSparkMaster().equals(SparkPipelineOptions.DEFAULT_MASTER_URL)) {
       sparkOptions.setSparkMaster(configuration.getSparkMasterUrl());
     }
+    if (sparkOptions.getSparkHistoryDir().equals(SparkPipelineOptions.DEFAULT_SPARK_HISTORY_DIR)) {
+      sparkOptions.setSparkHistoryDir(configuration.getSparkHistoryDir());
+    }
+    if (sparkOptions.getEventLogEnabled() == SparkPipelineOptions.DEFAULT_EVENT_LOG_ENABLED) {
+      sparkOptions.setEventLogEnabled(configuration.getEventLogEnabled());
+    }
 
     // Options can't be translated to proto if runner class is unresolvable, so set it to null.
     sparkOptions.setRunner(null);

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkJobServerDriver.java

@@ -26,6 +26,7 @@
 import org.kohsuke.args4j.CmdLineException;
 import org.kohsuke.args4j.CmdLineParser;
 import org.kohsuke.args4j.Option;
+import org.kohsuke.args4j.spi.ExplicitBooleanOptionHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -36,6 +37,27 @@ public class SparkJobServerDriver extends JobServerDriver {
 
   /** Spark runner-specific Configuration for the jobServer. */
   public static class SparkServerConfiguration extends ServerConfiguration {
+
+    @Option(
+        name = "--event-log-enabled",
+        usage = "Set it to true if event logs should be saved to Spark History Server directory",
+        handler = ExplicitBooleanOptionHandler.class)
+    private boolean eventLogEnabled = SparkPipelineOptions.DEFAULT_EVENT_LOG_ENABLED;
+
+    boolean getEventLogEnabled() {
+      return this.eventLogEnabled;
+    }
+
+    @Option(
+        name = "--spark-history-dir",
+        usage =
+            "Spark history dir path to store Spark History Server logs (e. g. /tmp/spark-events/)")
+    private String sparkHistoryDir = SparkPipelineOptions.DEFAULT_SPARK_HISTORY_DIR;
+
+    String getSparkHistoryDir() {
+      return this.sparkHistoryDir;
+    }
+
     @Option(
         name = "--spark-master-url",
         usage = "Spark master url to submit job (e.g. spark://host:port, local[4])")
@@ -72,7 +94,6 @@ private static SparkJobServerDriver fromParams(String[] args) {
       printUsage(parser);
       throw new IllegalArgumentException("Unable to parse command line arguments.", e);
     }
-
     return fromConfig(configuration);
   }
 

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.Boolean(false)
+  boolean getEventLogEnabled();
+
+  void setEventLogEnabled(boolean eventLogEnabled);
+
+  @Description("The directory to save Spark History Server logs")
+  @Default.String("/tmp/spark-events/")
+  String getSparkHistoryDir();
+
+  void setSparkHistoryDir(String sparkHistoryDir);
+
   @Description("Batch interval for Spark streaming in milliseconds.")
   @Default.Long(500)
   Long getBatchIntervalMillis();

runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineRunner.java

@@ -21,6 +21,8 @@
 import static org.apache.beam.runners.fnexecution.translation.PipelineTranslatorUtils.hasUnboundedPCollections;
 import static org.apache.beam.runners.spark.SparkPipelineOptions.prepareFilesToStage;
 
+import java.net.URI;
+import java.net.URISyntaxException;
 import java.util.UUID;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
@@ -42,6 +44,7 @@
 import org.apache.beam.runners.jobsubmission.PortablePipelineRunner;
 import org.apache.beam.runners.spark.aggregators.AggregatorsAccumulator;
 import org.apache.beam.runners.spark.metrics.MetricsAccumulator;
+import org.apache.beam.runners.spark.metrics.SparkBeamMetric;
 import org.apache.beam.runners.spark.translation.SparkBatchPortablePipelineTranslator;
 import org.apache.beam.runners.spark.translation.SparkContextFactory;
 import org.apache.beam.runners.spark.translation.SparkPortablePipelineTranslator;
@@ -56,22 +59,29 @@
 import org.apache.beam.vendor.grpc.v1p26p0.com.google.protobuf.Struct;
 import org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Preconditions;
 import org.apache.spark.api.java.JavaSparkContext;
+import org.apache.spark.scheduler.EventLoggingListener;
+import org.apache.spark.scheduler.SparkListenerApplicationEnd;
+import org.apache.spark.scheduler.SparkListenerApplicationStart;
+import org.apache.spark.scheduler.SparkListenerExecutorAdded;
+import org.apache.spark.scheduler.cluster.ExecutorInfo;
 import org.apache.spark.streaming.api.java.JavaStreamingContext;
 import org.apache.spark.streaming.api.java.JavaStreamingListener;
 import org.apache.spark.streaming.api.java.JavaStreamingListenerWrapper;
+import org.joda.time.Instant;
 import org.kohsuke.args4j.CmdLineException;
 import org.kohsuke.args4j.CmdLineParser;
 import org.kohsuke.args4j.Option;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import scala.Tuple2;
+import scala.collection.JavaConverters;
 
 /** Runs a portable pipeline on Apache Spark. */
 @SuppressWarnings({
   "rawtypes", // TODO(https://issues.apache.org/jira/browse/BEAM-10556)
   "nullness" // TODO(https://issues.apache.org/jira/browse/BEAM-10402)
 })
 public class SparkPipelineRunner implements PortablePipelineRunner {
-
   private static final Logger LOG = LoggerFactory.getLogger(SparkPipelineRunner.class);
 
   private final SparkPipelineOptions pipelineOptions;
@@ -81,7 +91,8 @@ public SparkPipelineRunner(SparkPipelineOptions pipelineOptions) {
   }
 
   @Override
-  public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo) {
+  public PortablePipelineResult run(RunnerApi.Pipeline pipeline, JobInfo jobInfo)
+      throws URISyntaxException {
     SparkPortablePipelineTranslator translator;
     boolean isStreaming = pipelineOptions.isStreaming() || hasUnboundedPCollections(pipeline);
     if (isStreaming) {
@@ -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(),
+                new ExecutorInfo(sparkMasters[i]._2(), 0, logUrlMap)));
+      }
+    }
+
     LOG.info(String.format("Running job %s on Spark master %s", jobInfo.jobId(), jsc.master()));
 
     // Initialize accumulators.
@@ -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(),
+              jsc.sparkUser(),
+              scala.Option.apply(jobInfo.jobName()),
+              scala.Option.apply(
+                  JavaConverters.mapAsScalaMapConverter(
+                          SparkBeamMetric.renderAllToString(result.metrics()))
+                      .asScala())));
+      eventLoggingListener.onApplicationEnd(
+          new SparkListenerApplicationEnd(Instant.now().getMillis()));
+      eventLoggingListener.stop();
+    }
 
     return result;
   }

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() {
+  static Map<String, ?> renderAll(MetricResults metricResults) {
     Map<String, Object> metrics = new HashMap<>();
-    MetricResults metricResults =
-        asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
     MetricQueryResults metricQueryResults = metricResults.allMetrics();
     for (MetricResult<Long> metricResult : metricQueryResults.getCounters()) {
       metrics.put(renderName(metricResult), metricResult.getAttempted());
@@ -63,8 +61,24 @@ class SparkBeamMetric implements Metric {
     return metrics;
   }
 
+  public static Map<String, String> renderAllToString(MetricResults metricResults) {
+    Map<String, String> metricsString = new HashMap<>();
+    for (Map.Entry<String, ?> entry : renderAll(metricResults).entrySet()) {
+      String key = entry.getKey();
+      String value = String.valueOf(entry.getValue());
+      metricsString.put(key, value);
+    }
+    return metricsString;
+  }
+
+  Map<String, ?> renderAll() {
+    MetricResults metricResults =
+        asAttemptedOnlyMetricResults(MetricsAccumulator.getInstance().value());
+    return renderAll(metricResults);
+  }
+
   @VisibleForTesting
-  String renderName(MetricResult<?> metricResult) {
+  static String renderName(MetricResult<?> metricResult) {
     MetricKey key = metricResult.getKey();
     MetricName name = key.metricName();
     String step = key.stepName();

runners/spark/src/test/java/org/apache/beam/runners/spark/metrics/SparkBeamMetricTest.java

@@ -35,7 +35,7 @@ public void testRenderName() {
                 "myStep.one.two(three)", MetricName.named("myNameSpace//", "myName()")),
             123,
             456);
-    String renderedName = new SparkBeamMetric().renderName(metricResult);
+    String renderedName = SparkBeamMetric.renderName(metricResult);
     assertThat(
         "Metric name was not rendered correctly",
         renderedName,