[proxima-beam-runner] init

datadrivencz · Mar 7, 2023 · 5f46464 · 5f46464
1 parent 9c5b340
commit 5f46464
Show file tree

Hide file tree

Showing 14 changed files with 839 additions and 0 deletions.
diff --git a/beam/pom.xml b/beam/pom.xml
@@ -199,6 +199,7 @@
     <module>core</module>
     <module>core-testing</module>
     <module>io-pubsub</module>
+    <module>runner</module>
     <module>tools</module>
   </modules>
 

diff --git a/beam/runner/license-header-spotless.txt b/beam/runner/license-header-spotless.txt
@@ -0,0 +1 @@
+../license-header-spotless.txt
diff --git a/beam/runner/license-header.txt b/beam/runner/license-header.txt
@@ -0,0 +1 @@
+../license-header.txt
diff --git a/beam/runner/pom.xml b/beam/runner/pom.xml
@@ -0,0 +1,207 @@
+<!--
+
+    Copyright 2017-2023 O2 Czech Republic, a.s.
+
+    Licensed 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.
+
+-->
+<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>
+
+  <parent>
+    <groupId>cz.o2.proxima</groupId>
+    <artifactId>proxima-beam</artifactId>
+    <version>0.11-SNAPSHOT</version>
+  </parent>
+
+  <artifactId>proxima-beam-runner</artifactId>
+  <packaging>jar</packaging>
+
+  <name>${project.groupId}:${project.artifactId}</name>
+
+  <build>
+    <extensions>
+      <extension>
+        <groupId>kr.motd.maven</groupId>
+        <artifactId>os-maven-plugin</artifactId>
+        <version>${maven.ext.os.version}</version>
+      </extension>
+    </extensions>
+
+    <plugins>
+
+      <!-- compile protobuffers -->
+      <plugin>
+        <groupId>org.xolstice.maven.plugins</groupId>
+        <artifactId>protobuf-maven-plugin</artifactId>
+        <version>${maven.protobuf.version}</version>
+        <configuration>
+          <protocArtifact>com.google.protobuf:protoc:${protoc.version}:exe:${os.detected.classifier}</protocArtifact>
+          <pluginId>grpc-java</pluginId>
+          <pluginArtifact>io.grpc:protoc-gen-grpc-java:${grpc.version}:exe:${os.detected.classifier}</pluginArtifact>
+        </configuration>
+        <executions>
+          <execution>
+            <goals>
+              <goal>compile</goal>
+              <goal>test-compile</goal>
+              <goal>compile-custom</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
+      <plugin>
+        <groupId>org.apache.maven.plugins</groupId>
+        <artifactId>maven-shade-plugin</artifactId>
+        <version>${maven.shade.version}</version>
+        <configuration>
+          <filters>
+            <filter>
+              <artifact>*:*</artifact>
+              <excludes>
+                <exclude>META-INF/*.SF</exclude>
+                <exclude>META-INF/*.DSA</exclude>
+                <exclude>META-INF/*.RSA</exclude>
+              </excludes>
+            </filter>
+          </filters>
+          <shadeTestJar>true</shadeTestJar>
+          <artifactSet>
+            <excludes>
+              <exclude>*:*</exclude>
+            </excludes>
+          </artifactSet>
+          <relocations>
+            <relocation>
+              <pattern>com.google.common.</pattern>
+              <shadedPattern>${coreShade}.com.google.common.</shadedPattern>
+            </relocation>
+          </relocations>
+        </configuration>
+        <executions>
+          <execution>
+            <phase>package</phase>
+            <goals>
+              <goal>shade</goal>
+            </goals>
+          </execution>
+        </executions>
+      </plugin>
+
+    </plugins>
+  </build>
+
+  <dependencies>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-core</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-direct-core</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-scheme-proto</artifactId>
+      <version>${project.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>${apache.beam.groupId}</groupId>
+      <artifactId>beam-sdks-java-core</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.beam</groupId>
+      <artifactId>beam-runners-java-job-service</artifactId>
+      <version>${apache.beam.version}</version>
+    </dependency>
+
+    <dependency>
+      <groupId>org.projectlombok</groupId>
+      <artifactId>lombok</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.auto.service</groupId>
+      <artifactId>auto-service</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>com.google.guava</groupId>
+      <artifactId>guava</artifactId>
+      <version>${guava.version}</version>
+      <scope>provided</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>org.apache.logging.log4j</groupId>
+      <artifactId>log4j-slf4j-impl</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-core</artifactId>
+      <version>${project.version}</version>
+      <classifier>tests</classifier>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-direct-core</artifactId>
+      <version>${project.version}</version>
+      <classifier>tests</classifier>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-direct-io-kafka</artifactId>
+      <version>${project.version}</version>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>cz.o2.proxima</groupId>
+      <artifactId>proxima-direct-io-kafka</artifactId>
+      <version>${project.version}</version>
+      <classifier>tests</classifier>
+      <scope>test</scope>
+    </dependency>
+
+    <dependency>
+      <groupId>junit</groupId>
+      <artifactId>junit</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.hamcrest</groupId>
+      <artifactId>hamcrest-all</artifactId>
+    </dependency>
+
+    <dependency>
+      <groupId>org.mockito</groupId>
+      <artifactId>mockito-core</artifactId>
+    </dependency>
+
+  </dependencies>
+
+</project>
+
diff --git a/beam/runner/src/main/java/cz/o2/proxima/beam/runner/ProximaBeamExecutor.java b/beam/runner/src/main/java/cz/o2/proxima/beam/runner/ProximaBeamExecutor.java
@@ -0,0 +1,73 @@
+/*
+ * Copyright 2017-2023 O2 Czech Republic, a.s.
+ *
+ * Licensed 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 cz.o2.proxima.beam.runner;
+
+import java.io.IOException;
+import org.apache.beam.model.jobmanagement.v1.JobApi;
+import org.apache.beam.model.pipeline.v1.RunnerApi.Pipeline;
+import org.apache.beam.runners.jobsubmission.PortablePipelineResult;
+import org.apache.beam.sdk.metrics.MetricResults;
+import org.checkerframework.checker.nullness.qual.NonNull;
+import org.joda.time.Duration;
+
+public class ProximaBeamExecutor {
+
+  private final Pipeline pipeline;
+  private final ProximaTranslationContext context;
+
+  ProximaBeamExecutor(Pipeline pipeline, ProximaTranslationContext context) {
+    this.pipeline = pipeline;
+    this.context = context;
+  }
+
+  public PortablePipelineResult execute() {
+    return asPipelineResult(context);
+  }
+
+  private PortablePipelineResult asPipelineResult(ProximaTranslationContext context) {
+    return new PortablePipelineResult() {
+      @Override
+      public JobApi.MetricResults portableMetrics() throws UnsupportedOperationException {
+        throw new UnsupportedOperationException();
+      }
+
+      @Override
+      public @NonNull State getState() {
+        return context.getState();
+      }
+
+      @Override
+      public @NonNull State cancel() throws IOException {
+        return context.cancel();
+      }
+
+      @Override
+      public @NonNull State waitUntilFinish(@NonNull Duration duration) {
+        return context.waitUntilFinish(duration);
+      }
+
+      @Override
+      public @NonNull State waitUntilFinish() {
+        return waitUntilFinish(Duration.ZERO);
+      }
+
+      @Override
+      public @NonNull MetricResults metrics() {
+        return context.getMetricsResult();
+      }
+    };
+  }
+}
diff --git a/beam/runner/src/main/java/cz/o2/proxima/beam/runner/ProximaPipelineOptions.java b/beam/runner/src/main/java/cz/o2/proxima/beam/runner/ProximaPipelineOptions.java
@@ -0,0 +1,52 @@
+/*
+ * Copyright 2017-2023 O2 Czech Republic, a.s.
+ *
+ * Licensed 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 cz.o2.proxima.beam.runner;
+
+import com.google.auto.service.AutoService;
+import java.util.Collections;
+import org.apache.beam.sdk.options.Default;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.options.PipelineOptionsRegistrar;
+import org.apache.beam.sdk.options.StreamingOptions;
+import org.checkerframework.checker.nullness.qual.NonNull;
+
+public interface ProximaPipelineOptions extends PipelineOptions, StreamingOptions {
+
+  @AutoService(PipelineOptionsRegistrar.class)
+  class OptionsFactory implements PipelineOptionsRegistrar {
+
+    @Override
+    public @NonNull Iterable<@NonNull Class<@NonNull ? extends @NonNull PipelineOptions>>
+        getPipelineOptions() {
+      return Collections.singletonList(ProximaPipelineOptions.class);
+    }
+  }
+
+  @Default.String("commit-log-family")
+  String getCommitFamily();
+
+  void setCommitFamily(String family);
+
+  @Default.String("state-random-access-family")
+  String getStateFamily();
+
+  void setStateFamily(String family);
+
+  @Default.String("shuffle-family")
+  String getShuffleFamily();
+
+  void setShuffleFamily(String family);
+}