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:
[email protected]
