You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nemo.apache.org by jo...@apache.org on 2019/03/18 06:51:15 UTC
[incubator-nemo] branch master updated: [NEMO-359] Implementation
of getEstimatedSizeBytes in SourceVertex (#202)
This is an automated email from the ASF dual-hosted git repository.
johnyangk pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/incubator-nemo.git
The following commit(s) were added to refs/heads/master by this push:
new 13f2e59 [NEMO-359] Implementation of getEstimatedSizeBytes in SourceVertex (#202)
13f2e59 is described below
commit 13f2e59b1dbff6ca27b7a064a41088c796724e29
Author: Won Wook SONG <wo...@apache.org>
AuthorDate: Mon Mar 18 15:51:11 2019 +0900
[NEMO-359] Implementation of getEstimatedSizeBytes in SourceVertex (#202)
JIRA: [NEMO-359: Implementation of getEstimatedSizeBytes in SourceVertex](https://issues.apache.org/jira/projects/NEMO/issues/NEMO-359)
**Major changes:**
- Addition of getEstimatedSizeBytes on the SourceVertex interface
- Implementation of the method for the bounded sources.
**Minor changes to note:**
- None
**Tests for the changes:**
- N/A
**Other comments:**
- None
Closes #202
---
.../src/main/java/org/apache/nemo/common/Util.java | 54 ++++++++++++++++++++++
.../nemo/common/ir/vertex/CachedSourceVertex.java | 16 ++++---
.../common/ir/vertex/InMemorySourceVertex.java | 11 ++++-
.../apache/nemo/common/ir/vertex/SourceVertex.java | 6 +++
.../apache/nemo/common/test/EmptyComponents.java | 9 +++-
.../beam/source/BeamBoundedSourceVertex.java | 22 +++++++--
.../beam/source/BeamUnboundedSourceVertex.java | 6 ++-
.../source/SparkDatasetBoundedSourceVertex.java | 12 ++++-
.../source/SparkTextFileBoundedSourceVertex.java | 9 +++-
.../nemo/runtime/common/metric/JobMetric.java | 8 ++--
.../runtime/executor/task/TaskExecutorTest.java | 7 ++-
.../nemo/runtime/master/metric/MetricStore.java | 2 +-
12 files changed, 138 insertions(+), 24 deletions(-)
diff --git a/common/src/main/java/org/apache/nemo/common/Util.java b/common/src/main/java/org/apache/nemo/common/Util.java
index 1d85820..4d9e042 100644
--- a/common/src/main/java/org/apache/nemo/common/Util.java
+++ b/common/src/main/java/org/apache/nemo/common/Util.java
@@ -18,6 +18,7 @@
*/
package org.apache.nemo.common;
+import org.apache.nemo.common.exception.MetricException;
import org.apache.nemo.common.ir.edge.IREdge;
import org.apache.nemo.common.ir.edge.executionproperty.*;
import org.apache.nemo.common.ir.vertex.IRVertex;
@@ -26,10 +27,16 @@ import org.apache.nemo.common.ir.vertex.utility.MessageBarrierVertex;
import org.apache.nemo.common.ir.vertex.utility.SamplingVertex;
import org.apache.nemo.common.ir.vertex.utility.StreamVertex;
+import java.io.IOException;
+import java.lang.instrument.Instrumentation;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.Paths;
import java.util.Collection;
import java.util.Optional;
import java.util.function.IntPredicate;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
/**
* Class to hold the utility methods.
@@ -38,6 +45,8 @@ public final class Util {
// Assume that this tag is never used in user application
public static final String CONTROL_EDGE_TAG = "CONTROL_EDGE";
+ private static Instrumentation instrumentation;
+
/**
* Private constructor for utility class.
*/
@@ -45,6 +54,33 @@ public final class Util {
}
/**
+ * Finds the project root path.
+ * @return the project root path.
+ */
+ public static String fetchProjectRootPath() {
+ return recursivelyFindLicense(Paths.get(System.getProperty("user.dir")));
+ }
+
+ /**
+ * Helper method to recursively find the LICENSE file.
+ * @param path the path to search for.
+ * @return the path containing the LICENSE file.
+ */
+ static String recursivelyFindLicense(final Path path) {
+ try (final Stream stream = Files.find(path, 1, (p, attributes) -> p.endsWith("LICENSE"))) {
+ if (stream.count() > 0) {
+ return path.toAbsolutePath().toString();
+ } else {
+ return recursivelyFindLicense(path.getParent());
+ }
+ } catch (NullPointerException e) {
+ return System.getProperty("user.dir");
+ } catch (IOException e) {
+ throw new MetricException(e);
+ }
+ }
+
+ /**
* Check the equality of two intPredicates.
* Check if the both the predicates either passes together or fails together for each
* integer in the range [0,noOfTimes]
@@ -165,4 +201,22 @@ public final class Util {
public static String stringifyIREdgeIds(final Collection<IREdge> edges) {
return edges.stream().map(IREdge::getId).sorted().collect(Collectors.toList()).toString();
}
+
+ /**
+ * Method for the instrumentation: for getting the object size.
+ * @param args arguments.
+ * @param inst the instrumentation.
+ */
+ public static void premain(final String args, final Instrumentation inst) {
+ instrumentation = inst;
+ }
+
+ /**
+ * Get the object byte size.
+ * @param o object to measure.
+ * @return the bytes of the object.
+ */
+ public static long getObjectSize(final Object o) {
+ return instrumentation.getObjectSize(o);
+ }
}
diff --git a/common/src/main/java/org/apache/nemo/common/ir/vertex/CachedSourceVertex.java b/common/src/main/java/org/apache/nemo/common/ir/vertex/CachedSourceVertex.java
index ecdd34b..331a635 100644
--- a/common/src/main/java/org/apache/nemo/common/ir/vertex/CachedSourceVertex.java
+++ b/common/src/main/java/org/apache/nemo/common/ir/vertex/CachedSourceVertex.java
@@ -47,17 +47,16 @@ public final class CachedSourceVertex<T> extends SourceVertex<T> {
/**
* Constructor for cloning.
*
- * @param readables the list of Readables to set.
+ * @param that The original CachedSourceVertex.
*/
- private CachedSourceVertex(final List<Readable<T>> readables) {
- this.readables = readables;
+ private CachedSourceVertex(final CachedSourceVertex<T> that) {
+ super(that);
+ this.readables = that.readables;
}
@Override
public CachedSourceVertex getClone() {
- final CachedSourceVertex that = new CachedSourceVertex<>(this.readables);
- this.copyExecutionPropertiesTo(that);
- return that;
+ return new CachedSourceVertex<>(this);
}
@Override
@@ -73,6 +72,11 @@ public final class CachedSourceVertex<T> extends SourceVertex<T> {
}
@Override
+ public long getEstimatedSizeBytes() {
+ return 0L;
+ }
+
+ @Override
public void clearInternalStates() {
readables = null;
}
diff --git a/common/src/main/java/org/apache/nemo/common/ir/vertex/InMemorySourceVertex.java b/common/src/main/java/org/apache/nemo/common/ir/vertex/InMemorySourceVertex.java
index 3278963..43936d8 100644
--- a/common/src/main/java/org/apache/nemo/common/ir/vertex/InMemorySourceVertex.java
+++ b/common/src/main/java/org/apache/nemo/common/ir/vertex/InMemorySourceVertex.java
@@ -20,6 +20,7 @@ package org.apache.nemo.common.ir.vertex;
import org.apache.nemo.common.ir.BoundedIteratorReadable;
import org.apache.nemo.common.ir.Readable;
+import org.apache.nemo.common.Util;
import java.io.IOException;
import java.util.ArrayList;
@@ -39,7 +40,6 @@ public final class InMemorySourceVertex<T> extends SourceVertex<T> {
* @param initializedSourceData the initial data object.
*/
public InMemorySourceVertex(final Iterable<T> initializedSourceData) {
- super();
this.initializedSourceData = initializedSourceData;
}
@@ -48,7 +48,7 @@ public final class InMemorySourceVertex<T> extends SourceVertex<T> {
*
* @param that the source object for copying
*/
- public InMemorySourceVertex(final InMemorySourceVertex that) {
+ private InMemorySourceVertex(final InMemorySourceVertex that) {
super(that);
this.initializedSourceData = that.initializedSourceData;
}
@@ -87,6 +87,13 @@ public final class InMemorySourceVertex<T> extends SourceVertex<T> {
}
@Override
+ public long getEstimatedSizeBytes() {
+ final ArrayList<Long> list = new ArrayList<>();
+ initializedSourceData.spliterator().forEachRemaining(obj -> list.add(Util.getObjectSize(obj)));
+ return list.stream().reduce((a, b) -> a + b).orElse(0L);
+ }
+
+ @Override
public void clearInternalStates() {
initializedSourceData = null;
}
diff --git a/common/src/main/java/org/apache/nemo/common/ir/vertex/SourceVertex.java b/common/src/main/java/org/apache/nemo/common/ir/vertex/SourceVertex.java
index 7ad6a82..f606cf5 100644
--- a/common/src/main/java/org/apache/nemo/common/ir/vertex/SourceVertex.java
+++ b/common/src/main/java/org/apache/nemo/common/ir/vertex/SourceVertex.java
@@ -59,6 +59,12 @@ public abstract class SourceVertex<O> extends IRVertex {
public abstract List<Readable<O>> getReadables(int desiredNumOfSplits) throws Exception;
/**
+ * Gets the estimated size of bytes. Returns 0L if not applicable.
+ * @return size of input bytes.
+ */
+ public abstract long getEstimatedSizeBytes();
+
+ /**
* Clears internal states, must be called after getReadables().
* Concretely, this clears the huge list of input splits held by objects like BeamBoundedSourceVertex before
* sending the vertex to remote executors.
diff --git a/common/src/main/java/org/apache/nemo/common/test/EmptyComponents.java b/common/src/main/java/org/apache/nemo/common/test/EmptyComponents.java
index 7282d41..45af7cf 100644
--- a/common/src/main/java/org/apache/nemo/common/test/EmptyComponents.java
+++ b/common/src/main/java/org/apache/nemo/common/test/EmptyComponents.java
@@ -211,8 +211,8 @@ public final class EmptyComponents {
*
* @param that the source object for copying
*/
- public EmptySourceVertex(final EmptySourceVertex that) {
- this.name = new String(that.name);
+ private EmptySourceVertex(final EmptySourceVertex that) {
+ this.name = that.name;
}
@Override
@@ -246,6 +246,11 @@ public final class EmptyComponents {
}
@Override
+ public long getEstimatedSizeBytes() {
+ return 0L;
+ }
+
+ @Override
public void clearInternalStates() {
}
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamBoundedSourceVertex.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamBoundedSourceVertex.java
index d0f4a0f..8d1f6a2 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamBoundedSourceVertex.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamBoundedSourceVertex.java
@@ -21,6 +21,7 @@ package org.apache.nemo.compiler.frontend.beam.source;
import com.fasterxml.jackson.databind.node.ObjectNode;
import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.nemo.common.exception.MetricException;
import org.apache.nemo.common.ir.Readable;
import java.io.IOException;
@@ -45,6 +46,7 @@ public final class BeamBoundedSourceVertex<O> extends SourceVertex<WindowedValue
private static final Logger LOG = LoggerFactory.getLogger(BeamBoundedSourceVertex.class.getName());
private BoundedSource<O> source;
private final DisplayData displayData;
+ private final long estimatedSizeBytes;
/**
* Constructor of BeamBoundedSourceVertex.
@@ -53,9 +55,13 @@ public final class BeamBoundedSourceVertex<O> extends SourceVertex<WindowedValue
* @param displayData data to display.
*/
public BeamBoundedSourceVertex(final BoundedSource<O> source, final DisplayData displayData) {
- super();
this.source = source;
this.displayData = displayData;
+ try {
+ this.estimatedSizeBytes = source.getEstimatedSizeBytes(null);
+ } catch (Exception e) {
+ throw new MetricException(e);
+ }
}
/**
@@ -63,10 +69,15 @@ public final class BeamBoundedSourceVertex<O> extends SourceVertex<WindowedValue
*
* @param that the source object for copying
*/
- public BeamBoundedSourceVertex(final BeamBoundedSourceVertex that) {
+ private BeamBoundedSourceVertex(final BeamBoundedSourceVertex that) {
super(that);
this.source = that.source;
this.displayData = that.displayData;
+ try {
+ this.estimatedSizeBytes = source.getEstimatedSizeBytes(null);
+ } catch (Exception e) {
+ throw new MetricException(e);
+ }
}
@Override
@@ -86,7 +97,7 @@ public final class BeamBoundedSourceVertex<O> extends SourceVertex<WindowedValue
if (source != null) {
LOG.info("estimate: {}", source.getEstimatedSizeBytes(null));
LOG.info("desired: {}", desiredNumOfSplits);
- source.split(source.getEstimatedSizeBytes(null) / desiredNumOfSplits, null)
+ source.split(this.estimatedSizeBytes / desiredNumOfSplits, null)
.forEach(boundedSource -> readables.add(new BoundedSourceReadable<>(boundedSource)));
return readables;
} else {
@@ -97,6 +108,11 @@ public final class BeamBoundedSourceVertex<O> extends SourceVertex<WindowedValue
}
@Override
+ public long getEstimatedSizeBytes() {
+ return this.estimatedSizeBytes;
+ }
+
+ @Override
public void clearInternalStates() {
source = null;
}
diff --git a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamUnboundedSourceVertex.java b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamUnboundedSourceVertex.java
index c2080cf..7371842 100644
--- a/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamUnboundedSourceVertex.java
+++ b/compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamUnboundedSourceVertex.java
@@ -55,7 +55,6 @@ public final class BeamUnboundedSourceVertex<O, M extends UnboundedSource.Checkp
*/
public BeamUnboundedSourceVertex(final UnboundedSource<O, M> source,
final DisplayData displayData) {
- super();
this.source = source;
this.displayData = displayData;
}
@@ -96,6 +95,11 @@ public final class BeamUnboundedSourceVertex<O, M extends UnboundedSource.Checkp
}
@Override
+ public long getEstimatedSizeBytes() {
+ return 0L;
+ }
+
+ @Override
public void clearInternalStates() {
source = null;
}
diff --git a/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkDatasetBoundedSourceVertex.java b/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkDatasetBoundedSourceVertex.java
index 31502be..d7b3794 100644
--- a/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkDatasetBoundedSourceVertex.java
+++ b/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkDatasetBoundedSourceVertex.java
@@ -37,6 +37,7 @@ import java.util.*;
*/
public final class SparkDatasetBoundedSourceVertex<T> extends SourceVertex<T> {
private List<Readable<T>> readables;
+ private long estimatedByteSize;
/**
* Constructor.
@@ -45,7 +46,6 @@ public final class SparkDatasetBoundedSourceVertex<T> extends SourceVertex<T> {
* @param dataset Dataset to read data from.
*/
public SparkDatasetBoundedSourceVertex(final SparkSession sparkSession, final Dataset<T> dataset) {
- super();
this.readables = new ArrayList<>();
final RDD rdd = dataset.sparkRDD();
final Partition[] partitions = rdd.getPartitions();
@@ -56,6 +56,9 @@ public final class SparkDatasetBoundedSourceVertex<T> extends SourceVertex<T> {
sparkSession.getInitialConf(),
i));
}
+ this.estimatedByteSize = dataset.javaRDD()
+ .map(o -> (long) o.toString().getBytes("UTF-8").length)
+ .reduce((a, b) -> a + b);
}
/**
@@ -63,7 +66,7 @@ public final class SparkDatasetBoundedSourceVertex<T> extends SourceVertex<T> {
*
* @param that the source object for copying
*/
- public SparkDatasetBoundedSourceVertex(final SparkDatasetBoundedSourceVertex<T> that) {
+ private SparkDatasetBoundedSourceVertex(final SparkDatasetBoundedSourceVertex<T> that) {
super(that);
this.readables = new ArrayList<>();
that.readables.forEach(this.readables::add);
@@ -85,6 +88,11 @@ public final class SparkDatasetBoundedSourceVertex<T> extends SourceVertex<T> {
}
@Override
+ public long getEstimatedSizeBytes() {
+ return this.estimatedByteSize;
+ }
+
+ @Override
public void clearInternalStates() {
readables = null;
}
diff --git a/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkTextFileBoundedSourceVertex.java b/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkTextFileBoundedSourceVertex.java
index f0294b1..5679c67 100644
--- a/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkTextFileBoundedSourceVertex.java
+++ b/compiler/frontend/spark/src/main/java/org/apache/nemo/compiler/frontend/spark/source/SparkTextFileBoundedSourceVertex.java
@@ -23,6 +23,7 @@ import org.apache.nemo.common.ir.Readable;
import org.apache.nemo.common.ir.vertex.SourceVertex;
import org.apache.spark.*;
import org.apache.spark.rdd.RDD;
+import org.apache.spark.util.SizeEstimator;
import scala.collection.JavaConverters;
import java.io.IOException;
@@ -33,6 +34,7 @@ import java.util.*;
*/
public final class SparkTextFileBoundedSourceVertex extends SourceVertex<String> {
private List<Readable<String>> readables;
+ private long estimatedSizeBytes;
/**
* Constructor.
@@ -44,7 +46,6 @@ public final class SparkTextFileBoundedSourceVertex extends SourceVertex<String>
public SparkTextFileBoundedSourceVertex(final SparkContext sparkContext,
final String inputPath,
final int numPartitions) {
- super();
this.readables = new ArrayList<>();
final Partition[] partitions = sparkContext.textFile(inputPath, numPartitions).getPartitions();
for (int i = 0; i < partitions.length; i++) {
@@ -55,6 +56,7 @@ public final class SparkTextFileBoundedSourceVertex extends SourceVertex<String>
inputPath,
numPartitions));
}
+ this.estimatedSizeBytes = SizeEstimator.estimate(sparkContext.textFile(inputPath, numPartitions));
}
/**
@@ -84,6 +86,11 @@ public final class SparkTextFileBoundedSourceVertex extends SourceVertex<String>
}
@Override
+ public long getEstimatedSizeBytes() {
+ return this.estimatedSizeBytes;
+ }
+
+ @Override
public void clearInternalStates() {
readables = null;
}
diff --git a/runtime/common/src/main/java/org/apache/nemo/runtime/common/metric/JobMetric.java b/runtime/common/src/main/java/org/apache/nemo/runtime/common/metric/JobMetric.java
index 5baea63..3b63cff 100644
--- a/runtime/common/src/main/java/org/apache/nemo/runtime/common/metric/JobMetric.java
+++ b/runtime/common/src/main/java/org/apache/nemo/runtime/common/metric/JobMetric.java
@@ -42,7 +42,7 @@ public final class JobMetric implements StateMetric<PlanState.State> {
private final String id;
private final List<StateTransitionEvent<PlanState.State>> stateTransitionEvents;
private String irDagSummary;
- private Integer inputSize;
+ private Long inputSize;
private String vertexProperties;
private String edgeProperties;
private JsonNode irDagJson;
@@ -74,7 +74,7 @@ public final class JobMetric implements StateMetric<PlanState.State> {
return this.irDagSummary;
}
- public Integer getInputSize() {
+ public Long getInputSize() {
return this.inputSize;
}
@@ -94,9 +94,9 @@ public final class JobMetric implements StateMetric<PlanState.State> {
this.irDagSummary = irDag.irDAGSummary();
this.inputSize = irDag.getRootVertices().stream()
.filter(irVertex -> irVertex instanceof SourceVertex)
- .mapToInt(irVertex -> {
+ .mapToLong(srcVertex -> {
try {
- return ((SourceVertex) irVertex).getReadables(1).size();
+ return ((SourceVertex) srcVertex).getEstimatedSizeBytes();
} catch (Exception e) {
throw new MetricException(e);
}
diff --git a/runtime/executor/src/test/java/org/apache/nemo/runtime/executor/task/TaskExecutorTest.java b/runtime/executor/src/test/java/org/apache/nemo/runtime/executor/task/TaskExecutorTest.java
index a0027e9..41dad41 100644
--- a/runtime/executor/src/test/java/org/apache/nemo/runtime/executor/task/TaskExecutorTest.java
+++ b/runtime/executor/src/test/java/org/apache/nemo/runtime/executor/task/TaskExecutorTest.java
@@ -643,7 +643,6 @@ public final class TaskExecutorTest {
* Source vertex for unbounded source test.
*/
private final class TestUnboundedSourceVertex extends SourceVertex {
-
@Override
public boolean isBounded() {
return false;
@@ -655,8 +654,12 @@ public final class TaskExecutorTest {
}
@Override
- public void clearInternalStates() {
+ public long getEstimatedSizeBytes() {
+ return 0L;
+ }
+ @Override
+ public void clearInternalStates() {
}
@Override
diff --git a/runtime/master/src/main/java/org/apache/nemo/runtime/master/metric/MetricStore.java b/runtime/master/src/main/java/org/apache/nemo/runtime/master/metric/MetricStore.java
index c4ad75f..76a8f52 100644
--- a/runtime/master/src/main/java/org/apache/nemo/runtime/master/metric/MetricStore.java
+++ b/runtime/master/src/main/java/org/apache/nemo/runtime/master/metric/MetricStore.java
@@ -281,7 +281,7 @@ public final class MetricStore {
final long duration = endTime - startTime; // ms
final String vertexProperties = jobMetric.getVertexProperties();
final String edgeProperties = jobMetric.getEdgeProperties();
- final Integer inputSize = jobMetric.getInputSize();
+ final Long inputSize = jobMetric.getInputSize();
final long jvmMemSize = Runtime.getRuntime().maxMemory();
final long memSize = ((com.sun.management.OperatingSystemMXBean) ManagementFactory
.getOperatingSystemMXBean()).getTotalPhysicalMemorySize();