add configuration options for flink platform bounded datastreams

Author: mspruc

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/mapping/BoundedTextFileSourceMapping.java

@@ -0,0 +1,55 @@
+/*
+ * 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.flink.mapping;
+
+import java.util.Collection;
+import java.util.Collections;
+
+import org.apache.wayang.basic.operators.TextFileSource;
+import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.core.mapping.OperatorPattern;
+import org.apache.wayang.core.mapping.PlanTransformation;
+import org.apache.wayang.core.mapping.ReplacementSubplanFactory;
+import org.apache.wayang.core.mapping.SubplanPattern;
+import org.apache.wayang.flink.operators.FlinkBoundedTextFileSource;
+import org.apache.wayang.flink.platform.FlinkPlatform;
+
+public class BoundedTextFileSourceMapping implements Mapping {
+    @Override
+    public Collection<PlanTransformation> getTransformations() {
+        return Collections.singleton(new PlanTransformation(
+                this.createSubplanPattern(),
+                this.createReplacementSubplanFactory(),
+                FlinkPlatform.getInstance()
+        ));
+    }
+
+    private SubplanPattern createSubplanPattern() {
+        final OperatorPattern<?> operatorPattern = new OperatorPattern<>(
+                "source", new TextFileSource("", null), false
+        );
+        return SubplanPattern.createSingleton(operatorPattern);
+    }
+
+    private ReplacementSubplanFactory createReplacementSubplanFactory() {
+        return new ReplacementSubplanFactory.OfSingleOperators<TextFileSource>(
+                (matchedOperator, epoch) -> new FlinkBoundedTextFileSource(matchedOperator).at(epoch)
+        );
+    }
+}

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/mapping/Mappings.java

@@ -19,16 +19,21 @@
 package org.apache.wayang.flink.mapping;
 
 import org.apache.wayang.core.mapping.Mapping;
+import org.apache.wayang.flink.plugin.FlinkBasicPlugin;
 
 import java.util.Arrays;
 import java.util.Collection;
 
 /**
- * Register for {@link Mapping}s for this platform.
+ * Register for {@link Mapping}s for {@link FlinkBasicPlugin}.
  */
 public class Mappings {
 
-    public static Collection<Mapping> BASIC_MAPPINGS = Arrays.asList(
+    /**
+     * Mappings using Flink's DataSets
+     * @deprecated DataSet API in Flink has been deprecated move over to bounded streams for a 1-to-1 replacement {@link #BOUNDED_STREAM_MAPPINGS}.
+     */
+    public static final Collection<Mapping> BASIC_MAPPINGS = Arrays.asList(
             new CartesianMapping(),
             new CoGroupMapping(),
             new CollectionSourceMapping(),
@@ -60,6 +65,9 @@ public class Mappings {
             new ZipWithIdMapping()
     );
 
+    public static final Collection<Mapping> BOUNDED_STREAM_MAPPINGS = Arrays.asList(
+        new BoundedTextFileSourceMapping()
+    );
 }
 
 

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/mapping/TextFileSourceMapping.java

@@ -33,7 +33,6 @@
 /**
  * Mapping from {@link TextFileSource} to {@link FlinkTextFileSource}.
  */
-@SuppressWarnings("unchecked")
 public class TextFileSourceMapping implements Mapping {
     @Override
     public Collection<PlanTransformation> getTransformations() {
@@ -46,7 +45,7 @@ public Collection<PlanTransformation> getTransformations() {
 
 
     private SubplanPattern createSubplanPattern() {
-        final OperatorPattern operatorPattern = new OperatorPattern(
+        final OperatorPattern<?> operatorPattern = new OperatorPattern<>(
                 "source", new TextFileSource("", null), false
         );
         return SubplanPattern.createSingleton(operatorPattern);

wayang-platforms/wayang-flink/src/main/java/org/apache/wayang/flink/plugin/FlinkBasicPlugin.java

@@ -35,15 +35,18 @@
 /**
  * This {@link Plugin} enables to use the basic Wayang {@link Operator}s on the {@link FlinkPlatform}.
  */
-public class FlinkBasicPlugin implements Plugin{
+public class FlinkBasicPlugin implements Plugin {
+    boolean useBoundedDataStreams = false;
+
     @Override
     public Collection<Platform> getRequiredPlatforms() {
         return Arrays.asList(FlinkPlatform.getInstance(), JavaPlatform.getInstance());
     }
 
     @Override
     public Collection<Mapping> getMappings() {
-        return Mappings.BASIC_MAPPINGS;
+
+        return useBoundedDataStreams ? Mappings.BOUNDED_STREAM_MAPPINGS : Mappings.BASIC_MAPPINGS;
     }
 
     @Override
@@ -53,6 +56,6 @@ public Collection<ChannelConversion> getChannelConversions() {
 
     @Override
     public void setProperties(Configuration configuration) {
-
+        useBoundedDataStreams = configuration.getBooleanProperty("wayang.flink.platforms.useDataStreams", false);
     }
 }