[NEMO-353] Launch NEXMark applications

README.md

@@ -95,6 +95,16 @@ Below describes how Beam applications can be run directly on Nemo.
  	-user_main org.apache.nemo.examples.beam.WordCount \
  	-optimization_policy org.apache.nemo.compiler.optimizer.policy.TransientResourcePolicy \
 	-user_args "hdfs://v-m:9000/test_input_wordcount hdfs://v-m:9000/test_output_wordcount"
+
+## NEXMark streaming Q0 (query0) example 
+./bin/run_nexmark.sh \
+ 	-job_id nexmark-Q0 \
+	-executor_json `pwd`/examples/resources/executors/beam_test_executor_resources.json \
+ 	-user_main org.apache.beam.sdk.nexmark.Main \
+ 	-optimization_policy org.apache.nemo.compiler.optimizer.policy.StreamingPolicy \
+  -scheduler_impl_class_name org.apache.nemo.runtime.master.scheduler.StreamingScheduler \	
+	-user_args "--runner=org.apache.nemo.client.beam.NemoRunner --streaming=true --query=0 --numEventGenerators=1"
+
 ```
 ## Resource Configuration
 `-executor_json` command line option can be used to provide a path to the JSON file that describes resource configuration for executors. Its default value is `config/default.json`, which initializes one of each `Transient`, `Reserved`, and `Compute` executor, each of which has one core and 1024MB memory.
@@ -148,3 +158,4 @@ Nemo Compiler and Engine can store JSON representation of intermediate DAGs.
 ## Speeding up builds 
 * To exclude Spark related packages: mvn clean install -T 2C -DskipTests -pl \\!compiler/frontend/spark,\\!examples/spark
 * To exclude Beam related packages: mvn clean install -T 2C -DskipTests -pl \\!compiler/frontend/beam,\\!examples/beam
+* To exclude NEXMark related packages: mvn clean install -T 2C -DskipTests -pl \\!examples/nexmark

bin/run_beam.sh

@@ -17,6 +17,8 @@
 # specific language governing permissions and limitations
 # under the License.
 
-java -Dlog4j.configuration=file://`pwd`/log4j.properties -cp examples/beam/target/nemo-examples-beam-$(mvn -q \
+VERSION=$(mvn -q \
   -Dexec.executable=echo -Dexec.args='${project.version}' \
-  --non-recursive exec:exec)-shaded.jar:`yarn classpath` org.apache.nemo.client.JobLauncher "$@"
+  --non-recursive exec:exec)
+
+java -Dlog4j.configuration=file://`pwd`/log4j.properties -cp client/target/nemo-client-$VERSION-shaded.jar:examples/beam/target/nemo-examples-beam-$VERSION-shaded.jar:`yarn classpath` org.apache.nemo.client.JobLauncher "$@"

bin/run_nexmark.sh

@@ -0,0 +1,24 @@
+#!/usr/bin/env bash
+#
+# 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.
+
+VERSION=$(mvn -q \
+  -Dexec.executable=echo -Dexec.args='${project.version}' \
+  --non-recursive exec:exec)
+
+java -Dlog4j.configuration=file://`pwd`/log4j.properties -cp client/target/nemo-client-$VERSION-shaded.jar:`yarn classpath`:examples/nexmark/target/nexmark-$VERSION-shaded.jar org.apache.nemo.client.JobLauncher "$@"

client/pom.xml

@@ -56,5 +56,56 @@ under the License.
             <artifactId>reef-runtime-yarn</artifactId>
             <version>${reef.version}</version>
         </dependency>
+
+
+      <!-- for nemo-beam-runner -->
+      <dependency>
+        <groupId>org.apache.nemo</groupId>
+        <artifactId>nemo-compiler-frontend-beam</artifactId>
+        <version>${project.version}</version>
+      </dependency>
+      <dependency>
+        <groupId>com.google.auto.service</groupId>
+        <artifactId>auto-service</artifactId>
+        <version>${auto-service.version}</version>
+        <optional>true</optional>
+      </dependency>
     </dependencies>
+
+
+  <build>
+    <plugins>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-deploy-plugin</artifactId>
+        <version>2.8.2</version>
+        <configuration>
+          <skip>true</skip>
+        </configuration>
+      </plugin>
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>3.0.0</version>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+            <configuration>
+              <outputFile>
+                ${project.build.directory}/${project.artifactId}-${project.version}-shaded.jar
+              </outputFile>
+              <transformers>
+                <!-- Required for using beam-hadoop: See https://stackoverflow.com/questions/44365545
+                -->
+                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer" />
+              </transformers>
+            </configuration>
+          </execution>
+        </executions>
+      </plugin>
+    </plugins>
+  </build>
 </project>

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/BeamStateTranslator.java → client/src/main/java/org/apache/nemo/client/beam/BeamStateTranslator.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.nemo.compiler.frontend.beam;
+package org.apache.nemo.client.beam;
 
 import org.apache.nemo.client.StateTranslator;
 import org.apache.nemo.runtime.common.state.PlanState;

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/NemoPipelineResult.java → client/src/main/java/org/apache/nemo/client/beam/NemoPipelineResult.java

@@ -16,7 +16,7 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.nemo.compiler.frontend.beam;
+package org.apache.nemo.client.beam;
 
 import org.apache.nemo.client.ClientEndpoint;
 import org.apache.beam.sdk.PipelineResult;

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/NemoRunner.java → client/src/main/java/org/apache/nemo/client/beam/NemoRunner.java

@@ -16,14 +16,16 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.nemo.compiler.frontend.beam;
+package org.apache.nemo.client.beam;
 
 import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.nemo.client.JobLauncher;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.PipelineRunner;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsValidator;
+import org.apache.nemo.compiler.frontend.beam.NemoPipelineOptions;
+import org.apache.nemo.compiler.frontend.beam.PipelineVisitor;
 
 /**
  * Runner class for BEAM programs.

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/NemoRunnerRegistrar.java → client/src/main/java/org/apache/nemo/client/beam/NemoRunnerRegistrar.java

@@ -16,14 +16,15 @@
  * specific language governing permissions and limitations
  * under the License.
  */
-package org.apache.nemo.compiler.frontend.beam;
+package org.apache.nemo.client.beam;
 
 import com.google.auto.service.AutoService;
 import com.google.common.collect.ImmutableList;
 import org.apache.beam.sdk.PipelineRunner;
 import org.apache.beam.sdk.options.PipelineOptions;
 import org.apache.beam.sdk.options.PipelineOptionsRegistrar;
 import org.apache.beam.sdk.runners.PipelineRunnerRegistrar;
+import org.apache.nemo.compiler.frontend.beam.NemoPipelineOptions;
 
 /**
  * Contains the {@link PipelineRunnerRegistrar} and {@link PipelineOptionsRegistrar} for the {@link NemoRunner}.

compiler/frontend/beam/pom.xml

@@ -7,9 +7,7 @@ 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
@@ -18,56 +16,51 @@ specific language governing permissions and limitations
 under the License.
 -->
 <project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
-    <modelVersion>4.0.0</modelVersion>
+  <modelVersion>4.0.0</modelVersion>
+
+  <parent>
+    <groupId>org.apache.nemo</groupId>
+    <artifactId>nemo-compiler</artifactId>
+    <version>0.2-SNAPSHOT</version>
+    <relativePath>../../</relativePath>
+  </parent>
+
+  <artifactId>nemo-compiler-frontend-beam</artifactId>
+  <name>Nemo Compiler Frontend: Beam</name>
+
+  <dependencies>
+    <dependency>
+      <groupId>org.apache.nemo</groupId>
+      <artifactId>nemo-common</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-sdks-java-core</artifactId>
+      <version>${beam.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-runners-core-java</artifactId>
+      <version>${beam.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-sdks-java-io-hadoop-input-format</artifactId>
+      <version>${beam.version}</version>
+    </dependency>
 
-    <parent>
-        <groupId>org.apache.nemo</groupId>
-        <artifactId>nemo-compiler</artifactId>
-        <version>0.2-SNAPSHOT</version>
-        <relativePath>../../</relativePath>
-    </parent>
+    <dependency>
+      <groupId>org.apache.hadoop</groupId>
+      <artifactId>hadoop-mapreduce-client-core</artifactId>
+      <version>${hadoop.version}</version>
+      <scope>provided</scope>
+    </dependency>
+    <!-- https://mvnrepository.com/artifact/com.amazonaws/aws-java-sdk -->
 
-    <artifactId>nemo-compiler-frontend-beam</artifactId>
-    <name>Nemo Compiler Frontend: Beam</name>
-    <packaging>jar</packaging>
 
-    <dependencies>
-	      <dependency>
-            <groupId>org.apache.nemo</groupId>
-            <artifactId>nemo-common</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.nemo</groupId>
-            <artifactId>nemo-client</artifactId>
-            <version>${project.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.beam</groupId>
-            <artifactId>beam-sdks-java-core</artifactId>
-            <version>${beam.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.beam</groupId>
-            <artifactId>beam-runners-core-java</artifactId>
-            <version>${beam.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.beam</groupId>
-            <artifactId>beam-sdks-java-io-hadoop-input-format</artifactId>
-            <version>${beam.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>org.apache.hadoop</groupId>
-            <artifactId>hadoop-mapreduce-client-core</artifactId>
-            <version>${hadoop.version}</version>
-            <scope>provided</scope>
-        </dependency>
-        <dependency>
-            <groupId>com.google.auto.service</groupId>
-            <artifactId>auto-service</artifactId>
-            <version>${auto-service.version}</version>
-            <optional>true</optional>
-        </dependency>
-    </dependencies>
+  </dependencies>
 </project>

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/InMemorySideInputReader.java

@@ -92,7 +92,7 @@ public void addSideInputElement(final PCollectionView<?> view,
 
   /**
    * Say a DoFn of this reader has 3 main inputs and 4 side inputs.
-   * {@link org.apache.nemo.runtime.executor.datatransfer.InputWatermarkManager} guarantees that the watermark here
+   * Nemo runtime guarantees that the watermark here
    * is the minimum of the all 7 input streams.
    * @param newWatermark to set.
    */

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/PipelineVisitor.java

@@ -35,7 +35,7 @@ public final class PipelineVisitor extends Pipeline.PipelineVisitor.Defaults {
    * @param pipeline to visit.
    * @param pipelineOptions pipeline options.
    */
-  PipelineVisitor(final Pipeline pipeline, final NemoPipelineOptions pipelineOptions) {
+  public PipelineVisitor(final Pipeline pipeline, final NemoPipelineOptions pipelineOptions) {
     this.context = new PipelineTranslationContext(pipeline, pipelineOptions);
   }
 
@@ -61,7 +61,7 @@ public void leaveCompositeTransform(final TransformHierarchy.Node node) {
   /**
    * @return the converted pipeline.
    */
-  IRDAG getConvertedPipeline() {
+  public IRDAG getConvertedPipeline() {
     return new IRDAG(context.getBuilder().build());
   }
 }

compiler/frontend/beam/src/main/java/org/apache/nemo/compiler/frontend/beam/source/BeamUnboundedSourceVertex.java

@@ -26,6 +26,7 @@
 import org.apache.nemo.common.ir.Readable;
 import org.apache.nemo.common.ir.vertex.IRVertex;
 import org.apache.nemo.common.ir.vertex.SourceVertex;
+import org.apache.nemo.common.test.EmptyComponents;
 import org.joda.time.Instant;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -81,10 +82,17 @@ public boolean isBounded() {
 
   @Override
   public List<Readable<Object>> getReadables(final int desiredNumOfSplits) throws Exception {
+
     final List<Readable<Object>> readables = new ArrayList<>();
-    source.split(desiredNumOfSplits, null)
-      .forEach(unboundedSource -> readables.add(new UnboundedSourceReadable<>(unboundedSource)));
-    return readables;
+    if (source != null) {
+      source.split(desiredNumOfSplits, null)
+        .forEach(unboundedSource -> readables.add(new UnboundedSourceReadable<>(unboundedSource)));
+      return readables;
+    } else {
+      // TODO #333: Remove SourceVertex#clearInternalStates
+      final SourceVertex emptySourceVertex = new EmptyComponents.EmptySourceVertex("EMPTY");
+      return emptySourceVertex.getReadables(desiredNumOfSplits);
+    }
   }
 
   @Override

examples/beam/pom.xml

@@ -36,6 +36,12 @@ under the License.
             <artifactId>nemo-compiler-frontend-beam</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nemo</groupId>
+            <artifactId>nemo-client</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>com.github.fommil.netlib</groupId>
             <artifactId>all</artifactId>

examples/beam/src/main/java/org/apache/nemo/examples/beam/AlternatingLeastSquare.java

@@ -20,12 +20,10 @@
 
 import com.github.fommil.netlib.BLAS;
 import com.github.fommil.netlib.LAPACK;
-import org.apache.nemo.compiler.frontend.beam.NemoRunner;
 import org.apache.nemo.compiler.frontend.beam.transform.LoopCompositeTransform;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.CoderProviders;
 import org.apache.beam.sdk.options.PipelineOptions;
-import org.apache.beam.sdk.options.PipelineOptionsFactory;
 import org.apache.beam.sdk.transforms.*;
 import org.apache.beam.sdk.values.KV;
 import org.apache.beam.sdk.values.PCollection;
@@ -353,7 +351,7 @@ public void processElement(final ProcessContext c) throws Exception {
    * Main function for the ALS BEAM program.
    * @param args arguments.
    */
-  public static void main(final String[] args) {
+  public static void main(final String[] args) throws ClassNotFoundException {
     final Long start = System.currentTimeMillis();
     LOG.info(Arrays.toString(args));
     final String inputFilePath = args[0];
@@ -374,8 +372,7 @@ public static void main(final String[] args) {
       outputFilePath = "";
     }
 
-    final PipelineOptions options = PipelineOptionsFactory.create();
-    options.setRunner(NemoRunner.class);
+    final PipelineOptions options = NemoPipelineOptionsFactory.create();
     options.setJobName("ALS");
     options.setStableUniqueNames(PipelineOptions.CheckEnabled.OFF);