Merge pull request #562 from juripetersen/logging-in-collect

Add explain to json config to Job

Author: zkaoudi

wayang-api/wayang-api-scala-java/src/main/scala/org/apache/wayang/api/PlanBuilder.scala

@@ -102,8 +102,13 @@ class PlanBuilder(private[api] val wayangContext: WayangContext, private var job
     */
   def buildAndExecute(): Unit = {
     val plan: WayangPlan = new WayangPlan(this.sinks.toArray: _*)
-    if (this.experiment == null) this.wayangContext.execute(jobName, plan, this.udfJars.toArray: _*)
-    else this.wayangContext.execute(jobName, plan, this.experiment, this.udfJars.toArray: _*)
+
+    if (this.experiment == null) {
+      this.wayangContext.execute(jobName, plan, this.udfJars.toArray: _*)
+    }
+    else {
+      this.wayangContext.execute(jobName, plan, this.experiment, this.udfJars.toArray: _*)
+    }
   }
 
   /**

wayang-commons/wayang-core/src/main/java/org/apache/wayang/core/api/Job.java

@@ -67,6 +67,7 @@
 import org.apache.wayang.core.util.OneTimeExecutable;
 import org.apache.wayang.core.util.ReflectionUtils;
 import org.apache.wayang.core.util.WayangCollections;
+import org.apache.wayang.core.util.ExplainUtils;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 
@@ -304,6 +305,16 @@ protected void doExecute() {
                 this.logger.warn("Failed to initialize monitor: {}", e);
             }
 
+            if (this.configuration.getBooleanProperty("wayang.core.explain.enabled")) {
+                ExplainUtils.write(
+                    ExplainUtils.parsePlan(this.wayangPlan, true),
+                    this.configuration.getStringProperty("wayang.core.explain.logical.file")
+                );
+                ExplainUtils.write(
+                    ExplainUtils.parsePlan(executionPlan, true),
+                    this.configuration.getStringProperty("wayang.core.explain.execution.file")
+                );
+            }
 
             // Take care of the execution.
             while (!this.execute(executionPlan, executionId)) {

wayang-commons/wayang-core/src/main/resources/wayang-core-defaults.properties

@@ -29,6 +29,7 @@ wayang.core.optimizer.enumeration.branchesfirst = false
 wayang.core.log.enabled = true
 # wayang.core.log.cardinalities = ~/.wayang/cardinalities.json
 # wayang.core.log.executions = ~/.wayang/executions.json
+wayang.core.explain.enabled = false
 
 # Configure re-optimization.
 wayang.core.optimizer.reoptimize = false