wayang-profiler/src/main/java/org/apache/wayang/profiler/log/LogEvaluator.java - incubator-wayang - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one
  * or more contributor license agreements.  See the NOTICE file
  * distributed with this work for additional information
  * regarding copyright ownership.  The ASF licenses this file
  * to you under the Apache License, Version 2.0 (the
  * "License"); you may not use this file except in compliance
  * with the License.  You may obtain a copy of the License at
  *
  *     http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */

 package org.apache.wayang.profiler.log;

 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStreamReader;
 import java.util.Collection;
 import java.util.Comparator;
 import java.util.LinkedList;
 import java.util.function.Predicate;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import org.apache.wayang.core.api.Configuration;
 import org.apache.wayang.core.api.exception.WayangException;
 import org.apache.wayang.core.optimizer.ProbabilisticIntervalEstimate;
 import org.apache.wayang.core.optimizer.costs.TimeEstimate;
 import org.apache.wayang.core.platform.AtomicExecutionGroup;
 import org.apache.wayang.core.platform.PartialExecution;
 import org.apache.wayang.core.profiling.ExecutionLog;
 import org.apache.wayang.core.util.Formats;
 import org.apache.wayang.java.platform.JavaPlatform;
 import org.apache.wayang.spark.platform.SparkPlatform;

 /**
  * Evaluates a {@link Configuration} on a {@link ExecutionLog}.
  */
 public class LogEvaluator {

     private static final Logger logger = LogManager.getLogger(LogEvaluator.class);

     private final Configuration configuration;

     private final Collection<PartialExecution> partialExecutions;

     private final Collection<Predicate<PartialExecution>> filters = new LinkedList<>();

     private Comparator<PartialExecution> sortCriterion;

     private boolean isSortAscending = true;

     public LogEvaluator(Configuration configuration) {
         // Initialize platforms - otherwise, we get some errors.
         JavaPlatform.getInstance();
         SparkPlatform.getInstance();

         this.configuration = configuration;
         this.partialExecutions = loadLog(configuration);

         // Print some general statistics.
         this.modifySorting("sort rel desc".split(" "));
         this.printStatistics();
     }

     private static Collection<PartialExecution> loadLog(Configuration configuration) {
         try (ExecutionLog executionLog = ExecutionLog.open(configuration)) {
             return executionLog.stream().collect(Collectors.toList());
         } catch (Exception e) {
             throw new WayangException("Could not evaluate execution log.", e);
         }
     }

     private void runUserLoop() throws IOException {
         String input;
         BufferedReader commandLine = new BufferedReader(new InputStreamReader(System.in));
         while ((input = commandLine.readLine()) != null) {
             final String[] tokens = input.split("\\s+");
             switch (tokens[0]) {
                 case "print":
                     this.printPartialExecutions(tokens);
                     break;
                 case "stats":
                     this.printStatistics();
                     break;
                 case "filter":
                     this.modifyFilters(tokens);
                     break;
                 case "sort":
                     this.modifySorting(tokens);
                     break;
                 case "exit":
                     return;
                 default:
                     System.out.println("Unknown command.");
             }
         }
     }

     private void printPartialExecutions(String[] commandLine) {
         Stream<PartialExecution> stream = createPartialExecutionStream();
         if (commandLine.length >= 2) {
             stream = stream.limit(Long.parseLong(commandLine[1]));
         }
         stream.forEach(this::print);
     }

     private void print(PartialExecution pe) {
         System.out.printf("Partial execution with %d execution groups:\n", pe.getAtomicExecutionGroups().size());
         System.out.printf("> Measured execution time: %s\n", Formats.formatDuration(pe.getMeasuredExecutionTime(), true));
         System.out.printf("> Estimated execution time: %s\n", pe.getOverallTimeEstimate(this.configuration));
         System.out.printf("> Delta: %s\n", pe.getOverallTimeEstimate(this.configuration).plus(-pe.getMeasuredExecutionTime()));
         for (AtomicExecutionGroup atomicExecutionGroup : pe.getAtomicExecutionGroups()) {
             System.out.printf("--> %s: %s\n", atomicExecutionGroup, atomicExecutionGroup.estimateExecutionTime());
         }
         System.out.println();
     }

     private void printStatistics() {
         // Print some general statistics.
         final TimeEstimate overallTimeEstimate = this.createPartialExecutionStream()
                 .map(partialExecution -> partialExecution.getOverallTimeEstimate(this.configuration))
                 .reduce(TimeEstimate.ZERO, TimeEstimate::plus);
         final long overallMeasuredTime = this.createPartialExecutionStream()
                 .map(PartialExecution::getMeasuredExecutionTime)
                 .reduce(0L, (a, b) -> a + b);
         System.out.printf("Found %d partial executions.\n", this.createPartialExecutionStream().count());
         System.out.printf("> Measured execution time: %s\n", Formats.formatDuration(overallMeasuredTime, true));
         System.out.printf("> Estimated execution time: %s\n", overallTimeEstimate);
         System.out.println();
     }

     private void modifyFilters(String[] commandLine) {
         if (commandLine.length == 1) {
             System.out.println("filter <name|platform|operator> <regex>");
             System.out.println("filter clear");
             return;
         }
         Predicate<PartialExecution> predicate;
         switch (commandLine[1]) {
             case "clear":
                 this.filters.clear();
                 return;
             case "name":
                 System.out.println("Not supported.");
 //                predicate = pe -> pe.getOperatorContexts().stream()
 //                        .map(operatorContext -> operatorContext.getOperator().getName())
 //                        .filter(Objects::nonNull)
 //                        .anyMatch(name -> matchSubstring(name, commandLine[2]));
                 break;
             case "operator":
                 System.out.println("Not supported.");
 //                predicate = pe -> pe.getOperatorContexts().stream()
 //                        .map(operatorContext -> operatorContext.getOperator().getClass().getSimpleName())
 //                        .anyMatch(name -> matchSubstring(name, commandLine[2]));
                 break;
             case "platform":
                 System.out.println("Not supported.");
 //                predicate = pe -> pe.getOperatorContexts().stream()
 //                        .map(operatorContext -> ((ExecutionOperator) operatorContext.getOperator()).getPlatform())
 //                        .anyMatch(platform -> matchSubstring(platform.getName(), commandLine[2]));
                 break;
             default:
                 System.out.println("Unknown filter type.");
                 return;
         }
 //        this.filters.add(predicate);
     }

     private void modifySorting(String[] commandLine) {
         if (commandLine.length == 1) {
             System.out.println("sort <est|run|delta> [asc|desc]");
             System.out.println("sort clear");
             return;
         }
         final Comparator<TimeEstimate> timeEstimateComparator = ProbabilisticIntervalEstimate.expectationValueComparator();
         switch (commandLine[1]) {
             case "clear":
                 this.sortCriterion = null;
                 return;
             case "est":
                 this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
                         a.getOverallTimeEstimate(this.configuration),
                         b.getOverallTimeEstimate(this.configuration)
                 );
                 break;
             case "run":
                 this.sortCriterion = (a, b) -> Long.compare(a.getMeasuredExecutionTime(), b.getMeasuredExecutionTime());
                 break;
             case "abs":
                 this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
                         a.getOverallTimeEstimate(this.configuration).plus(-a.getMeasuredExecutionTime()),
                         b.getOverallTimeEstimate(this.configuration).plus(-a.getMeasuredExecutionTime())
                 );
                 break;
             case "rel":
                 this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
                         a.getOverallTimeEstimate(this.configuration).times(1d / a.getMeasuredExecutionTime()),
                         b.getOverallTimeEstimate(this.configuration).times(1d / -a.getMeasuredExecutionTime())
                 );
                 break;
             default:
                 System.out.println("Unknown filter type.");
                 return;
         }
         this.isSortAscending = commandLine.length < 3 || commandLine[2].equalsIgnoreCase("asc");
     }

     private static boolean matchSubstring(String inputString, String regex) {
         return inputString.toLowerCase().matches(".*" + regex + ".*");
     }

     private Stream<PartialExecution> createPartialExecutionStream() {
         Stream<PartialExecution> stream = this.partialExecutions.stream();
         for (Predicate<PartialExecution> filter : this.filters) {
             stream = stream.filter(filter);
         }
         if (this.sortCriterion != null) {
             stream = stream.sorted(this.isSortAscending ?
                     this.sortCriterion :
                     (a, b) -> -this.sortCriterion.compare(a, b));
         }
         return stream;
     }


     public static void main(String[] args) throws IOException {
         LogEvaluator logEvaluator = new LogEvaluator(new Configuration());
         logEvaluator.runUserLoop();
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one
	* or more contributor license agreements. See the NOTICE file
	* distributed with this work for additional information
	* regarding copyright ownership. The ASF licenses this file
	* to you under the Apache License, Version 2.0 (the
	* "License"); you may not use this file except in compliance
	* with the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/

	package org.apache.wayang.profiler.log;

	import java.io.BufferedReader;
	import java.io.IOException;
	import java.io.InputStreamReader;
	import java.util.Collection;
	import java.util.Comparator;
	import java.util.LinkedList;
	import java.util.function.Predicate;
	import java.util.stream.Collectors;
	import java.util.stream.Stream;
	import org.apache.logging.log4j.LogManager;
	import org.apache.logging.log4j.Logger;
	import org.apache.wayang.core.api.Configuration;
	import org.apache.wayang.core.api.exception.WayangException;
	import org.apache.wayang.core.optimizer.ProbabilisticIntervalEstimate;
	import org.apache.wayang.core.optimizer.costs.TimeEstimate;
	import org.apache.wayang.core.platform.AtomicExecutionGroup;
	import org.apache.wayang.core.platform.PartialExecution;
	import org.apache.wayang.core.profiling.ExecutionLog;
	import org.apache.wayang.core.util.Formats;
	import org.apache.wayang.java.platform.JavaPlatform;
	import org.apache.wayang.spark.platform.SparkPlatform;

	/**
	* Evaluates a {@link Configuration} on a {@link ExecutionLog}.
	*/
	public class LogEvaluator {

	private static final Logger logger = LogManager.getLogger(LogEvaluator.class);

	private final Configuration configuration;

	private final Collection<PartialExecution> partialExecutions;

	private final Collection<Predicate<PartialExecution>> filters = new LinkedList<>();

	private Comparator<PartialExecution> sortCriterion;

	private boolean isSortAscending = true;

	public LogEvaluator(Configuration configuration) {
	// Initialize platforms - otherwise, we get some errors.
	JavaPlatform.getInstance();
	SparkPlatform.getInstance();

	this.configuration = configuration;
	this.partialExecutions = loadLog(configuration);

	// Print some general statistics.
	this.modifySorting("sort rel desc".split(" "));
	this.printStatistics();
	}

	private static Collection<PartialExecution> loadLog(Configuration configuration) {
	try (ExecutionLog executionLog = ExecutionLog.open(configuration)) {
	return executionLog.stream().collect(Collectors.toList());
	} catch (Exception e) {
	throw new WayangException("Could not evaluate execution log.", e);
	}
	}

	private void runUserLoop() throws IOException {
	String input;
	BufferedReader commandLine = new BufferedReader(new InputStreamReader(System.in));
	while ((input = commandLine.readLine()) != null) {
	final String[] tokens = input.split("\\s+");
	switch (tokens[0]) {
	case "print":
	this.printPartialExecutions(tokens);
	break;
	case "stats":
	this.printStatistics();
	break;
	case "filter":
	this.modifyFilters(tokens);
	break;
	case "sort":
	this.modifySorting(tokens);
	break;
	case "exit":
	return;
	default:
	System.out.println("Unknown command.");
	}
	}
	}

	private void printPartialExecutions(String[] commandLine) {
	Stream<PartialExecution> stream = createPartialExecutionStream();
	if (commandLine.length >= 2) {
	stream = stream.limit(Long.parseLong(commandLine[1]));
	}
	stream.forEach(this::print);
	}

	private void print(PartialExecution pe) {
	System.out.printf("Partial execution with %d execution groups:\n", pe.getAtomicExecutionGroups().size());
	System.out.printf("> Measured execution time: %s\n", Formats.formatDuration(pe.getMeasuredExecutionTime(), true));
	System.out.printf("> Estimated execution time: %s\n", pe.getOverallTimeEstimate(this.configuration));
	System.out.printf("> Delta: %s\n", pe.getOverallTimeEstimate(this.configuration).plus(-pe.getMeasuredExecutionTime()));
	for (AtomicExecutionGroup atomicExecutionGroup : pe.getAtomicExecutionGroups()) {
	System.out.printf("--> %s: %s\n", atomicExecutionGroup, atomicExecutionGroup.estimateExecutionTime());
	}
	System.out.println();
	}

	private void printStatistics() {
	// Print some general statistics.
	final TimeEstimate overallTimeEstimate = this.createPartialExecutionStream()
	.map(partialExecution -> partialExecution.getOverallTimeEstimate(this.configuration))
	.reduce(TimeEstimate.ZERO, TimeEstimate::plus);
	final long overallMeasuredTime = this.createPartialExecutionStream()
	.map(PartialExecution::getMeasuredExecutionTime)
	.reduce(0L, (a, b) -> a + b);
	System.out.printf("Found %d partial executions.\n", this.createPartialExecutionStream().count());
	System.out.printf("> Measured execution time: %s\n", Formats.formatDuration(overallMeasuredTime, true));
	System.out.printf("> Estimated execution time: %s\n", overallTimeEstimate);
	System.out.println();
	}

	private void modifyFilters(String[] commandLine) {
	if (commandLine.length == 1) {
	System.out.println("filter <name\|platform\|operator> <regex>");
	System.out.println("filter clear");
	return;
	}
	Predicate<PartialExecution> predicate;
	switch (commandLine[1]) {
	case "clear":
	this.filters.clear();
	return;
	case "name":
	System.out.println("Not supported.");
	// predicate = pe -> pe.getOperatorContexts().stream()
	// .map(operatorContext -> operatorContext.getOperator().getName())
	// .filter(Objects::nonNull)
	// .anyMatch(name -> matchSubstring(name, commandLine[2]));
	break;
	case "operator":
	System.out.println("Not supported.");
	// predicate = pe -> pe.getOperatorContexts().stream()
	// .map(operatorContext -> operatorContext.getOperator().getClass().getSimpleName())
	// .anyMatch(name -> matchSubstring(name, commandLine[2]));
	break;
	case "platform":
	System.out.println("Not supported.");
	// predicate = pe -> pe.getOperatorContexts().stream()
	// .map(operatorContext -> ((ExecutionOperator) operatorContext.getOperator()).getPlatform())
	// .anyMatch(platform -> matchSubstring(platform.getName(), commandLine[2]));
	break;
	default:
	System.out.println("Unknown filter type.");
	return;
	}
	// this.filters.add(predicate);
	}

	private void modifySorting(String[] commandLine) {
	if (commandLine.length == 1) {
	System.out.println("sort <est\|run\|delta> [asc\|desc]");
	System.out.println("sort clear");
	return;
	}
	final Comparator<TimeEstimate> timeEstimateComparator = ProbabilisticIntervalEstimate.expectationValueComparator();
	switch (commandLine[1]) {
	case "clear":
	this.sortCriterion = null;
	return;
	case "est":
	this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
	a.getOverallTimeEstimate(this.configuration),
	b.getOverallTimeEstimate(this.configuration)
	);
	break;
	case "run":
	this.sortCriterion = (a, b) -> Long.compare(a.getMeasuredExecutionTime(), b.getMeasuredExecutionTime());
	break;
	case "abs":
	this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
	a.getOverallTimeEstimate(this.configuration).plus(-a.getMeasuredExecutionTime()),
	b.getOverallTimeEstimate(this.configuration).plus(-a.getMeasuredExecutionTime())
	);
	break;
	case "rel":
	this.sortCriterion = (a, b) -> timeEstimateComparator.compare(
	a.getOverallTimeEstimate(this.configuration).times(1d / a.getMeasuredExecutionTime()),
	b.getOverallTimeEstimate(this.configuration).times(1d / -a.getMeasuredExecutionTime())
	);
	break;
	default:
	System.out.println("Unknown filter type.");
	return;
	}
	this.isSortAscending = commandLine.length < 3 \|\| commandLine[2].equalsIgnoreCase("asc");
	}

	private static boolean matchSubstring(String inputString, String regex) {
	return inputString.toLowerCase().matches("." + regex + ".");
	}

	private Stream<PartialExecution> createPartialExecutionStream() {
	Stream<PartialExecution> stream = this.partialExecutions.stream();
	for (Predicate<PartialExecution> filter : this.filters) {
	stream = stream.filter(filter);
	}
	if (this.sortCriterion != null) {
	stream = stream.sorted(this.isSortAscending ?
	this.sortCriterion :
	(a, b) -> -this.sortCriterion.compare(a, b));
	}
	return stream;
	}


	public static void main(String[] args) throws IOException {
	LogEvaluator logEvaluator = new LogEvaluator(new Configuration());
	logEvaluator.runUserLoop();
	}

	}