beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From k...@apache.org
Subject [34/39] incubator-beam git commit: BEAM-261 Make translators package private.
Date Sat, 12 Nov 2016 02:28:52 GMT
BEAM-261 Make translators package private.


Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/5553c603
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/5553c603
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/5553c603

Branch: refs/heads/master
Commit: 5553c603a0c48855d38d4702f19e905eac2034f2
Parents: 9197d1e
Author: Thomas Weise <thw@apache.org>
Authored: Thu Oct 27 16:19:15 2016 -0700
Committer: Thomas Weise <thw@apache.org>
Committed: Mon Nov 7 22:33:46 2016 +0100

----------------------------------------------------------------------
 .../runners/apex/ApexPipelineTranslator.java    | 185 -------
 .../apache/beam/runners/apex/ApexRunner.java    |  13 +-
 .../beam/runners/apex/ApexRunnerResult.java     |   2 +-
 .../translation/ApexPipelineTranslator.java     | 179 +++++++
 .../translation/CreateValuesTranslator.java     |  48 ++
 .../FlattenPCollectionTranslator.java           | 129 +++++
 .../apex/translation/GroupByKeyTranslator.java  |  42 ++
 .../translation/ParDoBoundMultiTranslator.java  | 142 ++++++
 .../apex/translation/ParDoBoundTranslator.java  |  64 +++
 .../translation/ReadUnboundedTranslator.java    |  42 ++
 .../apex/translation/TransformTranslator.java   |  31 ++
 .../apex/translation/TranslationContext.java    | 178 +++++++
 .../operators/ApexFlattenOperator.java          | 125 +++++
 .../operators/ApexGroupByKeyOperator.java       | 478 +++++++++++++++++++
 .../operators/ApexParDoOperator.java            | 375 +++++++++++++++
 .../ApexReadUnboundedInputOperator.java         | 155 ++++++
 .../translation/operators/package-info.java     |  22 +
 .../runners/apex/translation/package-info.java  |  22 +
 .../translation/utils/ApexStateInternals.java   | 438 +++++++++++++++++
 .../apex/translation/utils/ApexStreamTuple.java | 222 +++++++++
 .../utils/CoderAdapterStreamCodec.java          |  69 +++
 .../apex/translation/utils/NoOpStepContext.java |  72 +++
 .../utils/SerializablePipelineOptions.java      |  60 +++
 .../utils/ValueAndCoderKryoSerializable.java    |  77 +++
 .../apex/translation/utils/ValuesSource.java    | 149 ++++++
 .../apex/translation/utils/package-info.java    |  22 +
 .../translators/CreateValuesTranslator.java     |  48 --
 .../FlattenPCollectionTranslator.java           | 129 -----
 .../apex/translators/GroupByKeyTranslator.java  |  42 --
 .../translators/ParDoBoundMultiTranslator.java  | 142 ------
 .../apex/translators/ParDoBoundTranslator.java  |  64 ---
 .../translators/ReadUnboundedTranslator.java    |  42 --
 .../apex/translators/TransformTranslator.java   |  31 --
 .../apex/translators/TranslationContext.java    | 178 -------
 .../functions/ApexFlattenOperator.java          | 125 -----
 .../functions/ApexGroupByKeyOperator.java       | 478 -------------------
 .../functions/ApexParDoOperator.java            | 375 ---------------
 .../translators/functions/package-info.java     |  22 -
 .../io/ApexReadUnboundedInputOperator.java      | 154 ------
 .../apex/translators/io/ValuesSource.java       | 149 ------
 .../apex/translators/io/package-info.java       |  22 -
 .../runners/apex/translators/package-info.java  |  22 -
 .../translators/utils/ApexStateInternals.java   | 438 -----------------
 .../apex/translators/utils/ApexStreamTuple.java | 222 ---------
 .../utils/CoderAdapterStreamCodec.java          |  69 ---
 .../apex/translators/utils/NoOpStepContext.java |  72 ---
 .../utils/SerializablePipelineOptions.java      |  60 ---
 .../utils/ValueAndCoderKryoSerializable.java    |  77 ---
 .../apex/translators/utils/package-info.java    |  22 -
 .../translation/ApexGroupByKeyOperatorTest.java | 112 +++++
 .../FlattenPCollectionTranslatorTest.java       |  99 ++++
 .../translation/GroupByKeyTranslatorTest.java   | 246 ++++++++++
 .../translation/ParDoBoundTranslatorTest.java   | 340 +++++++++++++
 .../translation/ReadUnboundTranslatorTest.java  | 129 +++++
 .../utils/ApexStateInternalsTest.java           | 361 ++++++++++++++
 .../translation/utils/CollectionSource.java     | 136 ++++++
 .../translation/utils/PipelineOptionsTest.java  |  84 ++++
 .../translators/ApexGroupByKeyOperatorTest.java | 112 -----
 .../FlattenPCollectionTranslatorTest.java       |  99 ----
 .../translators/GroupByKeyTranslatorTest.java   | 246 ----------
 .../translators/ParDoBoundTranslatorTest.java   | 340 -------------
 .../translators/ReadUnboundTranslatorTest.java  | 129 -----
 .../utils/ApexStateInternalsTest.java           | 361 --------------
 .../translators/utils/CollectionSource.java     | 136 ------
 .../translators/utils/PipelineOptionsTest.java  |  84 ----
 65 files changed, 4653 insertions(+), 4685 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexPipelineTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexPipelineTranslator.java
deleted file mode 100644
index 8a87ce0..0000000
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexPipelineTranslator.java
+++ /dev/null
@@ -1,185 +0,0 @@
-/*
- * 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.beam.runners.apex;
-
-import com.datatorrent.api.DAG;
-
-import java.util.HashMap;
-import java.util.Map;
-
-import org.apache.beam.runners.apex.ApexRunner.CreateApexPCollectionView;
-import org.apache.beam.runners.apex.translators.CreateValuesTranslator;
-import org.apache.beam.runners.apex.translators.FlattenPCollectionTranslator;
-import org.apache.beam.runners.apex.translators.GroupByKeyTranslator;
-import org.apache.beam.runners.apex.translators.ParDoBoundMultiTranslator;
-import org.apache.beam.runners.apex.translators.ParDoBoundTranslator;
-import org.apache.beam.runners.apex.translators.ReadUnboundedTranslator;
-import org.apache.beam.runners.apex.translators.TransformTranslator;
-import org.apache.beam.runners.apex.translators.TranslationContext;
-import org.apache.beam.runners.apex.translators.io.ApexReadUnboundedInputOperator;
-import org.apache.beam.runners.core.UnboundedReadFromBoundedSource.BoundedToUnboundedSourceAdapter;
-import org.apache.beam.sdk.Pipeline;
-import org.apache.beam.sdk.io.Read;
-import org.apache.beam.sdk.runners.TransformTreeNode;
-import org.apache.beam.sdk.transforms.Create;
-import org.apache.beam.sdk.transforms.Flatten;
-import org.apache.beam.sdk.transforms.GroupByKey;
-import org.apache.beam.sdk.transforms.PTransform;
-import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.View.CreatePCollectionView;
-import org.apache.beam.sdk.values.PCollectionView;
-import org.apache.beam.sdk.values.PValue;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * {@link ApexPipelineTranslator} translates {@link Pipeline} objects
- * into Apex logical plan {@link DAG}.
- */
-@SuppressWarnings({"rawtypes", "unchecked"})
-public class ApexPipelineTranslator implements Pipeline.PipelineVisitor {
-  private static final Logger LOG = LoggerFactory.getLogger(ApexPipelineTranslator.class);
-
-  /**
-   * A map from {@link PTransform} subclass to the corresponding
-   * {@link TransformTranslator} to use to translate that transform.
-   */
-  private static final Map<Class<? extends PTransform>, TransformTranslator>
-      transformTranslators = new HashMap<>();
-
-  private final TranslationContext translationContext;
-
-  static {
-    // register TransformTranslators
-    registerTransformTranslator(ParDo.Bound.class, new ParDoBoundTranslator());
-    registerTransformTranslator(ParDo.BoundMulti.class, new ParDoBoundMultiTranslator<>());
-    registerTransformTranslator(Read.Unbounded.class, new ReadUnboundedTranslator());
-    registerTransformTranslator(Read.Bounded.class, new ReadBoundedTranslator());
-    registerTransformTranslator(GroupByKey.class, new GroupByKeyTranslator());
-    registerTransformTranslator(Flatten.FlattenPCollectionList.class,
-        new FlattenPCollectionTranslator());
-    registerTransformTranslator(Create.Values.class, new CreateValuesTranslator());
-    registerTransformTranslator(CreateApexPCollectionView.class,
-        new CreateApexPCollectionViewTranslator());
-    registerTransformTranslator(CreatePCollectionView.class,
-        new CreatePCollectionViewTranslator());
-  }
-
-  public ApexPipelineTranslator(TranslationContext translationContext) {
-    this.translationContext = translationContext;
-  }
-
-  public void translate(Pipeline pipeline) {
-    pipeline.traverseTopologically(this);
-  }
-
-  @Override
-  public CompositeBehavior enterCompositeTransform(TransformTreeNode node) {
-    LOG.debug("entering composite transform {}", node.getTransform());
-    return CompositeBehavior.ENTER_TRANSFORM;
-  }
-
-  @Override
-  public void leaveCompositeTransform(TransformTreeNode node) {
-    LOG.debug("leaving composite transform {}", node.getTransform());
-  }
-
-  @Override
-  public void visitPrimitiveTransform(TransformTreeNode node) {
-    LOG.debug("visiting transform {}", node.getTransform());
-    PTransform transform = node.getTransform();
-    TransformTranslator translator = getTransformTranslator(transform.getClass());
-    if (null == translator) {
-      throw new UnsupportedOperationException(
-          "no translator registered for " + transform);
-    }
-    translationContext.setCurrentTransform(node);
-    translator.translate(transform, translationContext);
-  }
-
-  @Override
-  public void visitValue(PValue value, TransformTreeNode producer) {
-    LOG.debug("visiting value {}", value);
-  }
-
-  /**
-   * Records that instances of the specified PTransform class
-   * should be translated by default by the corresponding
-   * {@link TransformTranslator}.
-   */
-  private static <TransformT extends PTransform> void registerTransformTranslator(
-      Class<TransformT> transformClass,
-      TransformTranslator<? extends TransformT> transformTranslator) {
-    if (transformTranslators.put(transformClass, transformTranslator) != null) {
-      throw new IllegalArgumentException(
-          "defining multiple translators for " + transformClass);
-    }
-  }
-
-  /**
-   * Returns the {@link TransformTranslator} to use for instances of the
-   * specified PTransform class, or null if none registered.
-   */
-  private <TransformT extends PTransform<?, ?>>
-  TransformTranslator<TransformT> getTransformTranslator(Class<TransformT> transformClass) {
-    return transformTranslators.get(transformClass);
-  }
-
-  private static class ReadBoundedTranslator<T> implements TransformTranslator<Read.Bounded<T>> {
-    private static final long serialVersionUID = 1L;
-
-    @Override
-    public void translate(Read.Bounded<T> transform, TranslationContext context) {
-      // TODO: adapter is visibleForTesting
-      BoundedToUnboundedSourceAdapter unboundedSource = new BoundedToUnboundedSourceAdapter<>(
-          transform.getSource());
-      ApexReadUnboundedInputOperator<T, ?> operator = new ApexReadUnboundedInputOperator<>(
-          unboundedSource, true, context.getPipelineOptions());
-      context.addOperator(operator, operator.output);
-    }
-
-  }
-
-  private static class CreateApexPCollectionViewTranslator<ElemT, ViewT>
-      implements TransformTranslator<CreateApexPCollectionView<ElemT, ViewT>> {
-    private static final long serialVersionUID = 1L;
-
-    @Override
-    public void translate(CreateApexPCollectionView<ElemT, ViewT> transform,
-        TranslationContext context) {
-      PCollectionView<ViewT> view = transform.getView();
-      context.addView(view);
-      LOG.debug("view {}", view.getName());
-    }
-  }
-
-  private static class CreatePCollectionViewTranslator<ElemT, ViewT>
-      implements TransformTranslator<CreatePCollectionView<ElemT, ViewT>> {
-    private static final long serialVersionUID = 1L;
-
-    @Override
-    public void translate(CreatePCollectionView<ElemT, ViewT> transform,
-        TranslationContext context) {
-      PCollectionView<ViewT> view = transform.getView();
-      context.addView(view);
-      LOG.debug("view {}", view.getName());
-    }
-  }
-
-}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
index 661308d..b42dddf 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunner.java
@@ -29,7 +29,7 @@ import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.List;
 
-import org.apache.beam.runners.apex.translators.TranslationContext;
+import org.apache.beam.runners.apex.translation.ApexPipelineTranslator;
 import org.apache.beam.runners.core.AssignWindows;
 import org.apache.beam.sdk.Pipeline;
 import org.apache.beam.sdk.coders.Coder;
@@ -118,17 +118,15 @@ public class ApexRunner extends PipelineRunner<ApexRunnerResult> {
   }
 
   @Override
-  public ApexRunnerResult run(Pipeline pipeline) {
+  public ApexRunnerResult run(final Pipeline pipeline) {
 
-    final TranslationContext translationContext = new TranslationContext(options);
-    ApexPipelineTranslator translator = new ApexPipelineTranslator(translationContext);
-    translator.translate(pipeline);
+    final ApexPipelineTranslator translator = new ApexPipelineTranslator(options);
 
     StreamingApplication apexApp = new StreamingApplication() {
       @Override
       public void populateDAG(DAG dag, Configuration conf) {
         dag.setAttribute(DAGContext.APPLICATION_NAME, options.getApplicationName());
-        translationContext.populateDAG(dag);
+        translator.translate(pipeline, dag);
       }
     };
 
@@ -352,9 +350,6 @@ public class ApexRunner extends PipelineRunner<ApexRunnerResult> {
 
   /**
    * Combiner that combines {@code T}s into a single {@code List<T>} containing all inputs.
-   *
-   * <p>For internal use by {@link StreamingViewAsMap}, {@link StreamingViewAsMultimap},
-   * {@link StreamingViewAsList}, {@link StreamingViewAsIterable}.
    * They require the input {@link PCollection} fits in memory.
    * For a large {@link PCollection} this is expected to crash!
    *

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerResult.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerResult.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerResult.java
index 03428a6..3ae69f2 100644
--- a/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerResult.java
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/ApexRunnerResult.java
@@ -79,7 +79,7 @@ public class ApexRunnerResult implements PipelineResult {
 
   /**
    * Return the DAG executed by the pipeline.
-   * @return
+   * @return DAG from translation.
    */
   public DAG getApexDAG() {
     return apexDAG;

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ApexPipelineTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ApexPipelineTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ApexPipelineTranslator.java
new file mode 100644
index 0000000..d38faf7
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ApexPipelineTranslator.java
@@ -0,0 +1,179 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import com.datatorrent.api.DAG;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.beam.runners.apex.ApexPipelineOptions;
+import org.apache.beam.runners.apex.ApexRunner.CreateApexPCollectionView;
+import org.apache.beam.runners.apex.translation.operators.ApexReadUnboundedInputOperator;
+import org.apache.beam.runners.core.UnboundedReadFromBoundedSource.BoundedToUnboundedSourceAdapter;
+import org.apache.beam.sdk.Pipeline;
+import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.runners.TransformTreeNode;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.transforms.View.CreatePCollectionView;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.PValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link ApexPipelineTranslator} translates {@link Pipeline} objects
+ * into Apex logical plan {@link DAG}.
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})
+public class ApexPipelineTranslator implements Pipeline.PipelineVisitor {
+  private static final Logger LOG = LoggerFactory.getLogger(ApexPipelineTranslator.class);
+
+  /**
+   * A map from {@link PTransform} subclass to the corresponding
+   * {@link TransformTranslator} to use to translate that transform.
+   */
+  private static final Map<Class<? extends PTransform>, TransformTranslator>
+      transformTranslators = new HashMap<>();
+
+  private final TranslationContext translationContext;
+
+  static {
+    // register TransformTranslators
+    registerTransformTranslator(ParDo.Bound.class, new ParDoBoundTranslator());
+    registerTransformTranslator(ParDo.BoundMulti.class, new ParDoBoundMultiTranslator<>());
+    registerTransformTranslator(Read.Unbounded.class, new ReadUnboundedTranslator());
+    registerTransformTranslator(Read.Bounded.class, new ReadBoundedTranslator());
+    registerTransformTranslator(GroupByKey.class, new GroupByKeyTranslator());
+    registerTransformTranslator(Flatten.FlattenPCollectionList.class,
+        new FlattenPCollectionTranslator());
+    registerTransformTranslator(Create.Values.class, new CreateValuesTranslator());
+    registerTransformTranslator(CreateApexPCollectionView.class,
+        new CreateApexPCollectionViewTranslator());
+    registerTransformTranslator(CreatePCollectionView.class,
+        new CreatePCollectionViewTranslator());
+  }
+
+  public ApexPipelineTranslator(ApexPipelineOptions options) {
+    this.translationContext = new TranslationContext(options);
+  }
+
+  public void translate(Pipeline pipeline, DAG dag) {
+    pipeline.traverseTopologically(this);
+    translationContext.populateDAG(dag);
+  }
+
+  @Override
+  public CompositeBehavior enterCompositeTransform(TransformTreeNode node) {
+    LOG.debug("entering composite transform {}", node.getTransform());
+    return CompositeBehavior.ENTER_TRANSFORM;
+  }
+
+  @Override
+  public void leaveCompositeTransform(TransformTreeNode node) {
+    LOG.debug("leaving composite transform {}", node.getTransform());
+  }
+
+  @Override
+  public void visitPrimitiveTransform(TransformTreeNode node) {
+    LOG.debug("visiting transform {}", node.getTransform());
+    PTransform transform = node.getTransform();
+    TransformTranslator translator = getTransformTranslator(transform.getClass());
+    if (null == translator) {
+      throw new UnsupportedOperationException(
+          "no translator registered for " + transform);
+    }
+    translationContext.setCurrentTransform(node);
+    translator.translate(transform, translationContext);
+  }
+
+  @Override
+  public void visitValue(PValue value, TransformTreeNode producer) {
+    LOG.debug("visiting value {}", value);
+  }
+
+  /**
+   * Records that instances of the specified PTransform class
+   * should be translated by default by the corresponding
+   * {@link TransformTranslator}.
+   */
+  private static <TransformT extends PTransform> void registerTransformTranslator(
+      Class<TransformT> transformClass,
+      TransformTranslator<? extends TransformT> transformTranslator) {
+    if (transformTranslators.put(transformClass, transformTranslator) != null) {
+      throw new IllegalArgumentException(
+          "defining multiple translators for " + transformClass);
+    }
+  }
+
+  /**
+   * Returns the {@link TransformTranslator} to use for instances of the
+   * specified PTransform class, or null if none registered.
+   */
+  private <TransformT extends PTransform<?, ?>>
+  TransformTranslator<TransformT> getTransformTranslator(Class<TransformT> transformClass) {
+    return transformTranslators.get(transformClass);
+  }
+
+  private static class ReadBoundedTranslator<T> implements TransformTranslator<Read.Bounded<T>> {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public void translate(Read.Bounded<T> transform, TranslationContext context) {
+      // TODO: adapter is visibleForTesting
+      BoundedToUnboundedSourceAdapter unboundedSource = new BoundedToUnboundedSourceAdapter<>(
+          transform.getSource());
+      ApexReadUnboundedInputOperator<T, ?> operator = new ApexReadUnboundedInputOperator<>(
+          unboundedSource, true, context.getPipelineOptions());
+      context.addOperator(operator, operator.output);
+    }
+
+  }
+
+  private static class CreateApexPCollectionViewTranslator<ElemT, ViewT>
+      implements TransformTranslator<CreateApexPCollectionView<ElemT, ViewT>> {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public void translate(CreateApexPCollectionView<ElemT, ViewT> transform,
+        TranslationContext context) {
+      PCollectionView<ViewT> view = transform.getView();
+      context.addView(view);
+      LOG.debug("view {}", view.getName());
+    }
+  }
+
+  private static class CreatePCollectionViewTranslator<ElemT, ViewT>
+      implements TransformTranslator<CreatePCollectionView<ElemT, ViewT>> {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public void translate(CreatePCollectionView<ElemT, ViewT> transform,
+        TranslationContext context) {
+      PCollectionView<ViewT> view = transform.getView();
+      context.addView(view);
+      LOG.debug("view {}", view.getName());
+    }
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/CreateValuesTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/CreateValuesTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/CreateValuesTranslator.java
new file mode 100644
index 0000000..ceae2b5
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/CreateValuesTranslator.java
@@ -0,0 +1,48 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import org.apache.beam.runners.apex.translation.operators.ApexReadUnboundedInputOperator;
+import org.apache.beam.runners.apex.translation.utils.ValuesSource;
+import org.apache.beam.sdk.coders.CannotProvideCoderException;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.transforms.Create;
+import org.apache.beam.sdk.values.PBegin;
+
+
+/**
+ * Wraps elements from Create.Values into an {@link UnboundedSource}.
+ * mainly used for testing
+ */
+class CreateValuesTranslator<T> implements TransformTranslator<Create.Values<T>> {
+  private static final long serialVersionUID = 1451000241832745629L;
+
+  @Override
+  public void translate(Create.Values<T> transform, TranslationContext context) {
+    try {
+      UnboundedSource<T, ?> unboundedSource = new ValuesSource<>(transform.getElements(),
+          transform.getDefaultOutputCoder((PBegin) context.getInput()));
+      ApexReadUnboundedInputOperator<T, ?> operator = new ApexReadUnboundedInputOperator<>(
+          unboundedSource, context.getPipelineOptions());
+      context.addOperator(operator, operator.output);
+    } catch (CannotProvideCoderException e) {
+      throw new RuntimeException(e);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/FlattenPCollectionTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/FlattenPCollectionTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/FlattenPCollectionTranslator.java
new file mode 100644
index 0000000..eb24af9
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/FlattenPCollectionTranslator.java
@@ -0,0 +1,129 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import com.google.common.collect.Lists;
+
+import java.util.Collections;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.beam.runners.apex.translation.operators.ApexFlattenOperator;
+import org.apache.beam.runners.apex.translation.operators.ApexReadUnboundedInputOperator;
+import org.apache.beam.runners.apex.translation.utils.ValuesSource;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.VoidCoder;
+import org.apache.beam.sdk.io.UnboundedSource;
+import org.apache.beam.sdk.transforms.Flatten;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionList;
+
+/**
+ * {@link Flatten.FlattenPCollectionList} translation to Apex operator.
+ */
+class FlattenPCollectionTranslator<T> implements
+    TransformTranslator<Flatten.FlattenPCollectionList<T>> {
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  public void translate(Flatten.FlattenPCollectionList<T> transform, TranslationContext context) {
+    PCollectionList<T> input = context.getInput();
+    List<PCollection<T>> collections = input.getAll();
+
+    if (collections.isEmpty()) {
+      // create a dummy source that never emits anything
+      @SuppressWarnings("unchecked")
+      UnboundedSource<T, ?> unboundedSource = new ValuesSource<>(Collections.EMPTY_LIST,
+          (Coder<T>) VoidCoder.of());
+      ApexReadUnboundedInputOperator<T, ?> operator = new ApexReadUnboundedInputOperator<>(
+          unboundedSource, context.getPipelineOptions());
+      context.addOperator(operator, operator.output);
+    } else {
+      PCollection<T> output = context.getOutput();
+      Map<PCollection<?>, Integer> unionTags = Collections.emptyMap();
+      flattenCollections(collections, unionTags, output, context);
+    }
+  }
+
+  /**
+   * Flatten the given collections into the given result collection. Translates
+   * into a cascading merge with 2 input ports per operator. The optional union
+   * tags can be used to identify the source in the result stream, used to
+   * channel multiple side inputs to a single Apex operator port.
+   *
+   * @param collections
+   * @param unionTags
+   * @param finalCollection
+   * @param context
+   */
+  static <T> void flattenCollections(List<PCollection<T>> collections, Map<PCollection<?>,
+      Integer> unionTags, PCollection<T> finalCollection, TranslationContext context) {
+    List<PCollection<T>> remainingCollections = Lists.newArrayList();
+    PCollection<T> firstCollection = null;
+    while (!collections.isEmpty()) {
+      for (PCollection<T> collection : collections) {
+        if (null == firstCollection) {
+          firstCollection = collection;
+        } else {
+          ApexFlattenOperator<T> operator = new ApexFlattenOperator<>();
+          context.addStream(firstCollection, operator.data1);
+          Integer unionTag = unionTags.get(firstCollection);
+          operator.data1Tag = (unionTag != null) ? unionTag : 0;
+          context.addStream(collection, operator.data2);
+          unionTag = unionTags.get(collection);
+          operator.data2Tag = (unionTag != null) ? unionTag : 0;
+
+          if (!collection.getCoder().equals(firstCollection.getCoder())) {
+              throw new UnsupportedOperationException("coders don't match");
+          }
+
+          if (collections.size() > 2) {
+            PCollection<T> intermediateCollection = intermediateCollection(collection,
+                collection.getCoder());
+            context.addOperator(operator, operator.out, intermediateCollection);
+            remainingCollections.add(intermediateCollection);
+          } else {
+            // final stream merge
+            context.addOperator(operator, operator.out, finalCollection);
+          }
+          firstCollection = null;
+        }
+      }
+      if (firstCollection != null) {
+        // push to next merge level
+        remainingCollections.add(firstCollection);
+        firstCollection = null;
+      }
+      if (remainingCollections.size() > 1) {
+        collections = remainingCollections;
+        remainingCollections = Lists.newArrayList();
+      } else {
+        collections = Lists.newArrayList();
+      }
+    }
+  }
+
+  static <T> PCollection<T> intermediateCollection(PCollection<T> input, Coder<T> outputCoder) {
+    PCollection<T> output = PCollection.createPrimitiveOutputInternal(input.getPipeline(),
+        input.getWindowingStrategy(), input.isBounded());
+    output.setCoder(outputCoder);
+    return output;
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/GroupByKeyTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/GroupByKeyTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/GroupByKeyTranslator.java
new file mode 100644
index 0000000..47d447a
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/GroupByKeyTranslator.java
@@ -0,0 +1,42 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import org.apache.beam.runners.apex.translation.operators.ApexGroupByKeyOperator;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+
+/**
+ * {@link GroupByKey} translation to Apex operator.
+ */
+class GroupByKeyTranslator<K, V> implements TransformTranslator<GroupByKey<K, V>> {
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  public void translate(GroupByKey<K, V> transform, TranslationContext context) {
+    PCollection<KV<K, V>> input = context.getInput();
+    ApexGroupByKeyOperator<K, V> group = new ApexGroupByKeyOperator<>(context.getPipelineOptions(),
+        input, context.<K>stateInternalsFactory()
+        );
+    context.addOperator(group, group.output);
+    context.addStream(input, group.input);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundMultiTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundMultiTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundMultiTranslator.java
new file mode 100644
index 0000000..7c91b91
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundMultiTranslator.java
@@ -0,0 +1,142 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.Operator.OutputPort;
+import com.google.common.collect.Maps;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.beam.runners.apex.translation.operators.ApexParDoOperator;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.util.WindowedValue.FullWindowedValueCoder;
+import org.apache.beam.sdk.util.WindowedValue.WindowedValueCoder;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionTuple;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * {@link ParDo.BoundMulti} is translated to {@link ApexParDoOperator} that wraps the {@link DoFn}.
+ */
+class ParDoBoundMultiTranslator<InputT, OutputT>
+    implements TransformTranslator<ParDo.BoundMulti<InputT, OutputT>> {
+  private static final long serialVersionUID = 1L;
+  private static final Logger LOG = LoggerFactory.getLogger(ParDoBoundMultiTranslator.class);
+
+  @Override
+  public void translate(ParDo.BoundMulti<InputT, OutputT> transform, TranslationContext context) {
+    OldDoFn<InputT, OutputT> doFn = transform.getFn();
+    PCollectionTuple output = context.getOutput();
+    PCollection<InputT> input = context.getInput();
+    List<PCollectionView<?>> sideInputs = transform.getSideInputs();
+    Coder<InputT> inputCoder = input.getCoder();
+    WindowedValueCoder<InputT> wvInputCoder = FullWindowedValueCoder.of(inputCoder,
+        input.getWindowingStrategy().getWindowFn().windowCoder());
+
+    ApexParDoOperator<InputT, OutputT> operator = new ApexParDoOperator<>(
+        context.getPipelineOptions(),
+        doFn, transform.getMainOutputTag(), transform.getSideOutputTags().getAll(),
+        context.<PCollection<?>>getInput().getWindowingStrategy(), sideInputs, wvInputCoder,
+        context.<Void>stateInternalsFactory()
+        );
+
+    Map<TupleTag<?>, PCollection<?>> outputs = output.getAll();
+    Map<PCollection<?>, OutputPort<?>> ports = Maps.newHashMapWithExpectedSize(outputs.size());
+    for (Map.Entry<TupleTag<?>, PCollection<?>> outputEntry : outputs.entrySet()) {
+      if (outputEntry.getKey() == transform.getMainOutputTag()) {
+        ports.put(outputEntry.getValue(), operator.output);
+      } else {
+        int portIndex = 0;
+        for (TupleTag<?> tag : transform.getSideOutputTags().getAll()) {
+          if (tag == outputEntry.getKey()) {
+            ports.put(outputEntry.getValue(), operator.sideOutputPorts[portIndex]);
+            break;
+          }
+          portIndex++;
+        }
+      }
+    }
+    context.addOperator(operator, ports);
+    context.addStream(context.getInput(), operator.input);
+    if (!sideInputs.isEmpty()) {
+      addSideInputs(operator, sideInputs, context);
+    }
+  }
+
+  static void addSideInputs(ApexParDoOperator<?, ?> operator, List<PCollectionView<?>> sideInputs,
+      TranslationContext context) {
+    Operator.InputPort<?>[] sideInputPorts = {operator.sideInput1};
+    if (sideInputs.size() > sideInputPorts.length) {
+      PCollection<?> unionCollection = unionSideInputs(sideInputs, context);
+      context.addStream(unionCollection, sideInputPorts[0]);
+    } else {
+      // the number of ports for side inputs is fixed and each port can only take one input.
+      for (int i = 0; i < sideInputs.size(); i++) {
+        context.addStream(context.getViewInput(sideInputs.get(i)), sideInputPorts[i]);
+      }
+    }
+  }
+
+  private static PCollection<?> unionSideInputs(List<PCollectionView<?>> sideInputs,
+      TranslationContext context) {
+    checkArgument(sideInputs.size() > 1, "requires multiple side inputs");
+    // flatten and assign union tag
+    List<PCollection<Object>> sourceCollections = new ArrayList<>();
+    Map<PCollection<?>, Integer> unionTags = new HashMap<>();
+    PCollection<Object> firstSideInput = context.getViewInput(sideInputs.get(0));
+    for (int i = 0; i < sideInputs.size(); i++) {
+      PCollectionView<?> sideInput = sideInputs.get(i);
+      PCollection<?> sideInputCollection = context.getViewInput(sideInput);
+      if (!sideInputCollection.getWindowingStrategy().equals(
+          firstSideInput.getWindowingStrategy())) {
+        // TODO: check how to handle this in stream codec
+        //String msg = "Multiple side inputs with different window strategies.";
+        //throw new UnsupportedOperationException(msg);
+        LOG.warn("Side inputs union with different windowing strategies {} {}",
+            firstSideInput.getWindowingStrategy(), sideInputCollection.getWindowingStrategy());
+      }
+      if (!sideInputCollection.getCoder().equals(firstSideInput.getCoder())) {
+        String msg = "Multiple side inputs with different coders.";
+        throw new UnsupportedOperationException(msg);
+      }
+      sourceCollections.add(context.<PCollection<Object>>getViewInput(sideInput));
+      unionTags.put(sideInputCollection, i);
+    }
+
+    PCollection<Object> resultCollection = FlattenPCollectionTranslator.intermediateCollection(
+        firstSideInput, firstSideInput.getCoder());
+    FlattenPCollectionTranslator.flattenCollections(sourceCollections, unionTags, resultCollection,
+        context);
+    return resultCollection;
+
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslator.java
new file mode 100644
index 0000000..c1ebbd5
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ParDoBoundTranslator.java
@@ -0,0 +1,64 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import java.util.List;
+
+import org.apache.beam.runners.apex.translation.operators.ApexParDoOperator;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.transforms.DoFn;
+import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.ParDo;
+import org.apache.beam.sdk.util.WindowedValue.FullWindowedValueCoder;
+import org.apache.beam.sdk.util.WindowedValue.WindowedValueCoder;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.apache.beam.sdk.values.TupleTagList;
+
+/**
+ * {@link ParDo.Bound} is translated to {link ApexParDoOperator} that wraps the {@link DoFn}.
+ */
+class ParDoBoundTranslator<InputT, OutputT> implements
+    TransformTranslator<ParDo.Bound<InputT, OutputT>> {
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  public void translate(ParDo.Bound<InputT, OutputT> transform, TranslationContext context) {
+    OldDoFn<InputT, OutputT> doFn = transform.getFn();
+    PCollection<OutputT> output = context.getOutput();
+    PCollection<InputT> input = context.getInput();
+    List<PCollectionView<?>> sideInputs = transform.getSideInputs();
+    Coder<InputT> inputCoder = input.getCoder();
+    WindowedValueCoder<InputT> wvInputCoder = FullWindowedValueCoder.of(inputCoder,
+        input.getWindowingStrategy().getWindowFn().windowCoder());
+
+    ApexParDoOperator<InputT, OutputT> operator = new ApexParDoOperator<>(
+        context.getPipelineOptions(),
+        doFn, new TupleTag<OutputT>(), TupleTagList.empty().getAll() /*sideOutputTags*/,
+        output.getWindowingStrategy(), sideInputs, wvInputCoder,
+        context.<Void>stateInternalsFactory()
+        );
+    context.addOperator(operator, operator.output);
+    context.addStream(context.getInput(), operator.input);
+    if (!sideInputs.isEmpty()) {
+       ParDoBoundMultiTranslator.addSideInputs(operator, sideInputs, context);
+    }
+  }
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ReadUnboundedTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ReadUnboundedTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ReadUnboundedTranslator.java
new file mode 100644
index 0000000..b3034ac
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/ReadUnboundedTranslator.java
@@ -0,0 +1,42 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import com.datatorrent.api.InputOperator;
+
+import org.apache.beam.runners.apex.translation.operators.ApexReadUnboundedInputOperator;
+import org.apache.beam.sdk.io.Read;
+import org.apache.beam.sdk.io.UnboundedSource;
+
+/**
+ * {@link Read.Unbounded} is translated to Apex {@link InputOperator}
+ * that wraps {@link UnboundedSource}.
+ */
+class ReadUnboundedTranslator<T> implements TransformTranslator<Read.Unbounded<T>> {
+  private static final long serialVersionUID = 1L;
+
+  @Override
+  public void translate(Read.Unbounded<T> transform, TranslationContext context) {
+    UnboundedSource<T, ?> unboundedSource = transform.getSource();
+    ApexReadUnboundedInputOperator<T, ?> operator = new ApexReadUnboundedInputOperator<>(
+        unboundedSource, context.getPipelineOptions());
+    context.addOperator(operator, operator.output);
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TransformTranslator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TransformTranslator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TransformTranslator.java
new file mode 100644
index 0000000..eb81052
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TransformTranslator.java
@@ -0,0 +1,31 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+
+import java.io.Serializable;
+
+import org.apache.beam.sdk.transforms.PTransform;
+
+/**
+ * Translates {@link PTransform} to Apex functions.
+ */
+interface TransformTranslator<T extends PTransform<?, ?>> extends Serializable {
+  void translate(T transform, TranslationContext context);
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TranslationContext.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TranslationContext.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TranslationContext.java
new file mode 100644
index 0000000..e016730
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/TranslationContext.java
@@ -0,0 +1,178 @@
+/*
+ * 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.beam.runners.apex.translation;
+
+import static com.google.common.base.Preconditions.checkArgument;
+
+import com.datatorrent.api.Context.PortContext;
+import com.datatorrent.api.DAG;
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.Operator.InputPort;
+import com.datatorrent.api.Operator.OutputPort;
+
+import java.util.ArrayList;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+import org.apache.beam.runners.apex.ApexPipelineOptions;
+import org.apache.beam.runners.apex.translation.utils.ApexStateInternals;
+import org.apache.beam.runners.apex.translation.utils.ApexStreamTuple;
+import org.apache.beam.runners.apex.translation.utils.CoderAdapterStreamCodec;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.runners.TransformTreeNode;
+import org.apache.beam.sdk.transforms.AppliedPTransform;
+import org.apache.beam.sdk.transforms.PTransform;
+import org.apache.beam.sdk.util.WindowedValue.FullWindowedValueCoder;
+import org.apache.beam.sdk.util.state.StateInternalsFactory;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.PInput;
+import org.apache.beam.sdk.values.POutput;
+import org.apache.commons.lang3.tuple.ImmutablePair;
+import org.apache.commons.lang3.tuple.Pair;
+
+/**
+ * Maintains context data for {@link TransformTranslator}s.
+ */
+@SuppressWarnings({"rawtypes", "unchecked"})
+class TranslationContext {
+
+  private final ApexPipelineOptions pipelineOptions;
+  private AppliedPTransform<?, ?, ?> currentTransform;
+  private final Map<PCollection, Pair<OutputPort<?>, List<InputPort<?>>>> streams = new HashMap<>();
+  private final Map<String, Operator> operators = new HashMap<>();
+  private final Map<PCollectionView<?>, PInput> viewInputs = new HashMap<>();
+
+  public void addView(PCollectionView<?> view) {
+    this.viewInputs.put(view, this.getInput());
+  }
+
+  public <InputT extends PInput> InputT getViewInput(PCollectionView<?> view) {
+    PInput input = this.viewInputs.get(view);
+    checkArgument(input != null, "unknown view " + view.getName());
+    return (InputT) input;
+  }
+
+  TranslationContext(ApexPipelineOptions pipelineOptions) {
+    this.pipelineOptions = pipelineOptions;
+  }
+
+  public void setCurrentTransform(TransformTreeNode treeNode) {
+    this.currentTransform = AppliedPTransform.of(treeNode.getFullName(),
+        treeNode.getInput(), treeNode.getOutput(), (PTransform) treeNode.getTransform());
+  }
+
+  public ApexPipelineOptions getPipelineOptions() {
+    return pipelineOptions;
+  }
+
+  public <InputT extends PInput> InputT getInput() {
+    return (InputT) getCurrentTransform().getInput();
+  }
+
+  public <OutputT extends POutput> OutputT getOutput() {
+    return (OutputT) getCurrentTransform().getOutput();
+  }
+
+  private AppliedPTransform<?, ?, ?> getCurrentTransform() {
+    checkArgument(currentTransform != null, "current transform not set");
+    return currentTransform;
+  }
+
+  public void addOperator(Operator operator, OutputPort port) {
+    addOperator(operator, port, this.<PCollection<?>>getOutput());
+  }
+
+  /**
+   * Register operator and output ports for the given collections.
+   * @param operator
+   * @param ports
+   */
+  public void addOperator(Operator operator, Map<PCollection<?>, OutputPort<?>> ports) {
+    boolean first = true;
+    for (Map.Entry<PCollection<?>, OutputPort<?>> portEntry : ports.entrySet()) {
+      if (first) {
+        addOperator(operator, portEntry.getValue(), portEntry.getKey());
+        first = false;
+      } else {
+        this.streams.put(portEntry.getKey(), (Pair) new ImmutablePair<>(portEntry.getValue(),
+            new ArrayList<>()));
+      }
+    }
+  }
+
+  /**
+   * Add the operator with its output port for the given result {link PCollection}.
+   * @param operator
+   * @param port
+   * @param output
+   */
+  public void addOperator(Operator operator, OutputPort port, PCollection output) {
+    // Apex DAG requires a unique operator name
+    // use the transform's name and make it unique
+    String name = getCurrentTransform().getFullName();
+    for (int i = 1; this.operators.containsKey(name); i++) {
+      name = getCurrentTransform().getFullName() + i;
+    }
+    this.operators.put(name, operator);
+    this.streams.put(output, (Pair) new ImmutablePair<>(port, new ArrayList<>()));
+  }
+
+  public void addStream(PInput input, InputPort inputPort) {
+    Pair<OutputPort<?>, List<InputPort<?>>> stream = this.streams.get(input);
+    checkArgument(stream != null, "no upstream operator defined for %s", input);
+    stream.getRight().add(inputPort);
+  }
+
+  public void populateDAG(DAG dag) {
+    for (Map.Entry<String, Operator> nameAndOperator : this.operators.entrySet()) {
+      dag.addOperator(nameAndOperator.getKey(), nameAndOperator.getValue());
+    }
+    int streamIndex = 0;
+    for (Map.Entry<PCollection, Pair<OutputPort<?>, List<InputPort<?>>>> streamEntry : this.
+        streams.entrySet()) {
+      List<InputPort<?>> sinksList = streamEntry.getValue().getRight();
+      InputPort[] sinks = sinksList.toArray(new InputPort[sinksList.size()]);
+      if (sinks.length > 0) {
+        dag.addStream("stream" + streamIndex++, streamEntry.getValue().getLeft(), sinks);
+        for (InputPort port : sinks) {
+          PCollection pc = streamEntry.getKey();
+          Coder coder = pc.getCoder();
+          if (pc.getWindowingStrategy() != null) {
+            coder = FullWindowedValueCoder.of(pc.getCoder(),
+                pc.getWindowingStrategy().getWindowFn().windowCoder()
+                );
+          }
+          Coder<Object> wrapperCoder = ApexStreamTuple.ApexStreamTupleCoder.of(coder);
+          CoderAdapterStreamCodec streamCodec = new CoderAdapterStreamCodec(wrapperCoder);
+          dag.setInputPortAttribute(port, PortContext.STREAM_CODEC, streamCodec);
+        }
+      }
+    }
+  }
+
+  /**
+   * Return the {@link StateInternalsFactory} for the pipeline translation.
+   * @return
+   */
+  public <K> StateInternalsFactory<K> stateInternalsFactory() {
+    return new ApexStateInternals.ApexStateInternalsFactory();
+  }
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexFlattenOperator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexFlattenOperator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexFlattenOperator.java
new file mode 100644
index 0000000..3d9db51
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexFlattenOperator.java
@@ -0,0 +1,125 @@
+/*
+ * 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.beam.runners.apex.translation.operators;
+
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.datatorrent.common.util.BaseOperator;
+
+import org.apache.beam.runners.apex.translation.utils.ApexStreamTuple;
+import org.apache.beam.runners.apex.translation.utils.ApexStreamTuple.WatermarkTuple;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Apex operator for Beam {@link org.apache.beam.sdk.transforms.Flatten.FlattenPCollectionList}.
+ */
+public class ApexFlattenOperator<InputT> extends BaseOperator {
+
+  private static final Logger LOG = LoggerFactory.getLogger(ApexFlattenOperator.class);
+  private boolean traceTuples = false;
+
+  private long inputWM1;
+  private long inputWM2;
+  private long outputWM;
+
+  public int data1Tag;
+  public int data2Tag;
+
+  /**
+   * Data input port 1.
+   */
+  public final transient DefaultInputPort<ApexStreamTuple<WindowedValue<InputT>>> data1 =
+      new DefaultInputPort<ApexStreamTuple<WindowedValue<InputT>>>() {
+    /**
+     * Emits to port "out"
+     */
+    @Override
+    public void process(ApexStreamTuple<WindowedValue<InputT>> tuple) {
+      if (tuple instanceof WatermarkTuple) {
+        WatermarkTuple<?> wmTuple = (WatermarkTuple<?>) tuple;
+        if (wmTuple.getTimestamp() > inputWM1) {
+          inputWM1 = wmTuple.getTimestamp();
+          if (inputWM1 <= inputWM2) {
+            // move output watermark and emit it
+            outputWM = inputWM1;
+            if (traceTuples) {
+              LOG.debug("\nemitting watermark {}\n", outputWM);
+            }
+            out.emit(tuple);
+          }
+        }
+        return;
+      }
+      if (traceTuples) {
+        LOG.debug("\nemitting {}\n", tuple);
+      }
+
+      if (data1Tag > 0 && tuple instanceof ApexStreamTuple.DataTuple) {
+        ((ApexStreamTuple.DataTuple<?>) tuple).setUnionTag(data1Tag);
+      }
+      out.emit(tuple);
+    }
+  };
+
+  /**
+   * Data input port 2.
+   */
+  public final transient DefaultInputPort<ApexStreamTuple<WindowedValue<InputT>>> data2 =
+      new DefaultInputPort<ApexStreamTuple<WindowedValue<InputT>>>() {
+    /**
+     * Emits to port "out"
+     */
+    @Override
+    public void process(ApexStreamTuple<WindowedValue<InputT>> tuple) {
+      if (tuple instanceof WatermarkTuple) {
+        WatermarkTuple<?> wmTuple = (WatermarkTuple<?>) tuple;
+        if (wmTuple.getTimestamp() > inputWM2) {
+          inputWM2 = wmTuple.getTimestamp();
+          if (inputWM2 <= inputWM1) {
+            // move output watermark and emit it
+            outputWM = inputWM2;
+            if (traceTuples) {
+              LOG.debug("\nemitting watermark {}\n", outputWM);
+            }
+            out.emit(tuple);
+          }
+        }
+        return;
+      }
+      if (traceTuples) {
+        LOG.debug("\nemitting {}\n", tuple);
+      }
+
+      if (data2Tag > 0 && tuple instanceof ApexStreamTuple.DataTuple) {
+        ((ApexStreamTuple.DataTuple<?>) tuple).setUnionTag(data2Tag);
+      }
+      out.emit(tuple);
+    }
+  };
+
+  /**
+   * Output port.
+   */
+  @OutputPortFieldAnnotation(optional = true)
+  public final transient DefaultOutputPort<ApexStreamTuple<WindowedValue<InputT>>> out =
+    new DefaultOutputPort<>();
+
+}

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/5553c603/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
----------------------------------------------------------------------
diff --git a/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
new file mode 100644
index 0000000..1b5e693
--- /dev/null
+++ b/runners/apex/src/main/java/org/apache/beam/runners/apex/translation/operators/ApexGroupByKeyOperator.java
@@ -0,0 +1,478 @@
+/*
+ * 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.beam.runners.apex.translation.operators;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+import com.datatorrent.api.Context.OperatorContext;
+import com.datatorrent.api.DefaultInputPort;
+import com.datatorrent.api.DefaultOutputPort;
+import com.datatorrent.api.Operator;
+import com.datatorrent.api.StreamCodec;
+import com.datatorrent.api.annotation.OutputPortFieldAnnotation;
+import com.esotericsoftware.kryo.serializers.FieldSerializer.Bind;
+import com.esotericsoftware.kryo.serializers.JavaSerializer;
+import com.google.common.base.Throwables;
+import com.google.common.collect.HashMultimap;
+import com.google.common.collect.Multimap;
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.Set;
+
+import org.apache.beam.runners.apex.ApexPipelineOptions;
+import org.apache.beam.runners.apex.translation.utils.ApexStreamTuple;
+import org.apache.beam.runners.apex.translation.utils.SerializablePipelineOptions;
+import org.apache.beam.runners.core.GroupAlsoByWindowViaWindowSetDoFn;
+import org.apache.beam.runners.core.SystemReduceFn;
+import org.apache.beam.sdk.coders.Coder;
+import org.apache.beam.sdk.coders.CoderException;
+import org.apache.beam.sdk.coders.KvCoder;
+import org.apache.beam.sdk.options.PipelineOptions;
+import org.apache.beam.sdk.transforms.Aggregator;
+import org.apache.beam.sdk.transforms.Combine;
+import org.apache.beam.sdk.transforms.GroupByKey;
+import org.apache.beam.sdk.transforms.OldDoFn;
+import org.apache.beam.sdk.transforms.windowing.BoundedWindow;
+import org.apache.beam.sdk.transforms.windowing.PaneInfo;
+import org.apache.beam.sdk.util.CoderUtils;
+import org.apache.beam.sdk.util.KeyedWorkItem;
+import org.apache.beam.sdk.util.KeyedWorkItems;
+import org.apache.beam.sdk.util.TimeDomain;
+import org.apache.beam.sdk.util.TimerInternals;
+import org.apache.beam.sdk.util.WindowedValue;
+import org.apache.beam.sdk.util.WindowingInternals;
+import org.apache.beam.sdk.util.WindowingStrategy;
+import org.apache.beam.sdk.util.state.StateInternals;
+import org.apache.beam.sdk.util.state.StateInternalsFactory;
+import org.apache.beam.sdk.util.state.StateNamespace;
+import org.apache.beam.sdk.values.KV;
+import org.apache.beam.sdk.values.PCollection;
+import org.apache.beam.sdk.values.PCollectionView;
+import org.apache.beam.sdk.values.TupleTag;
+import org.joda.time.Instant;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+/**
+ * Apex operator for Beam {@link GroupByKey}.
+ * This operator expects the input stream already partitioned by K,
+ * which is determined by the {@link StreamCodec} on the input port.
+ *
+ * @param <K> key type
+ * @param <V> value type
+ */
+public class ApexGroupByKeyOperator<K, V> implements Operator {
+  private static final Logger LOG = LoggerFactory.getLogger(ApexGroupByKeyOperator.class);
+  private boolean traceTuples = true;
+
+  @Bind(JavaSerializer.class)
+  private WindowingStrategy<V, BoundedWindow> windowingStrategy;
+  @Bind(JavaSerializer.class)
+  private Coder<K> keyCoder;
+  @Bind(JavaSerializer.class)
+  private Coder<V> valueCoder;
+
+  @Bind(JavaSerializer.class)
+  private final SerializablePipelineOptions serializedOptions;
+  @Bind(JavaSerializer.class)
+  private final StateInternalsFactory<K> stateInternalsFactory;
+  private Map<ByteBuffer, StateInternals<K>> perKeyStateInternals = new HashMap<>();
+  private Map<ByteBuffer, Set<TimerInternals.TimerData>> activeTimers = new HashMap<>();
+
+  private transient ProcessContext context;
+  private transient OldDoFn<KeyedWorkItem<K, V>, KV<K, Iterable<V>>> fn;
+  private transient ApexTimerInternals timerInternals = new ApexTimerInternals();
+  private Instant inputWatermark = BoundedWindow.TIMESTAMP_MIN_VALUE;
+
+  public final transient DefaultInputPort<ApexStreamTuple<WindowedValue<KV<K, V>>>> input =
+      new DefaultInputPort<ApexStreamTuple<WindowedValue<KV<K, V>>>>() {
+    @Override
+    public void process(ApexStreamTuple<WindowedValue<KV<K, V>>> t) {
+      try {
+        if (t instanceof ApexStreamTuple.WatermarkTuple) {
+          ApexStreamTuple.WatermarkTuple<?> mark = (ApexStreamTuple.WatermarkTuple<?>) t;
+          processWatermark(mark);
+          if (traceTuples) {
+            LOG.debug("\nemitting watermark {}\n", mark.getTimestamp());
+          }
+          output.emit(ApexStreamTuple.WatermarkTuple.<WindowedValue<KV<K, Iterable<V>>>>of(
+              mark.getTimestamp()));
+          return;
+        }
+        if (traceTuples) {
+          LOG.debug("\ninput {}\n", t.getValue());
+        }
+        processElement(t.getValue());
+      } catch (Exception e) {
+        Throwables.propagateIfPossible(e);
+        throw new RuntimeException(e);
+      }
+    }
+  };
+
+  @OutputPortFieldAnnotation(optional = true)
+  public final transient DefaultOutputPort<ApexStreamTuple<WindowedValue<KV<K, Iterable<V>>>>>
+      output = new DefaultOutputPort<>();
+
+  @SuppressWarnings("unchecked")
+  public ApexGroupByKeyOperator(ApexPipelineOptions pipelineOptions, PCollection<KV<K, V>> input,
+      StateInternalsFactory<K> stateInternalsFactory) {
+    checkNotNull(pipelineOptions);
+    this.serializedOptions = new SerializablePipelineOptions(pipelineOptions);
+    this.windowingStrategy = (WindowingStrategy<V, BoundedWindow>) input.getWindowingStrategy();
+    this.keyCoder = ((KvCoder<K, V>) input.getCoder()).getKeyCoder();
+    this.valueCoder = ((KvCoder<K, V>) input.getCoder()).getValueCoder();
+    this.stateInternalsFactory = stateInternalsFactory;
+  }
+
+  @SuppressWarnings("unused") // for Kryo
+  private ApexGroupByKeyOperator() {
+    this.serializedOptions = null;
+    this.stateInternalsFactory = null;
+  }
+
+  @Override
+  public void beginWindow(long l) {
+  }
+
+  @Override
+  public void endWindow() {
+  }
+
+  @Override
+  public void setup(OperatorContext context) {
+    this.traceTuples = ApexStreamTuple.Logging.isDebugEnabled(serializedOptions.get(), this);
+    StateInternalsFactory<K> stateInternalsFactory = new GroupByKeyStateInternalsFactory();
+    this.fn = GroupAlsoByWindowViaWindowSetDoFn.create(this.windowingStrategy,
+        stateInternalsFactory, SystemReduceFn.<K, V, BoundedWindow>buffering(this.valueCoder));
+    this.context = new ProcessContext(fn, this.timerInternals);
+  }
+
+  @Override
+  public void teardown() {
+  }
+
+  /**
+   * Returns the list of timers that are ready to fire. These are the timers
+   * that are registered to be triggered at a time before the current watermark.
+   * We keep these timers in a Set, so that they are deduplicated, as the same
+   * timer can be registered multiple times.
+   */
+  private Multimap<ByteBuffer, TimerInternals.TimerData> getTimersReadyToProcess(
+      long currentWatermark) {
+
+    // we keep the timers to return in a different list and launch them later
+    // because we cannot prevent a trigger from registering another trigger,
+    // which would lead to concurrent modification exception.
+    Multimap<ByteBuffer, TimerInternals.TimerData> toFire = HashMultimap.create();
+
+    Iterator<Map.Entry<ByteBuffer, Set<TimerInternals.TimerData>>> it =
+        activeTimers.entrySet().iterator();
+    while (it.hasNext()) {
+      Map.Entry<ByteBuffer, Set<TimerInternals.TimerData>> keyWithTimers = it.next();
+
+      Iterator<TimerInternals.TimerData> timerIt = keyWithTimers.getValue().iterator();
+      while (timerIt.hasNext()) {
+        TimerInternals.TimerData timerData = timerIt.next();
+        if (timerData.getTimestamp().isBefore(currentWatermark)) {
+          toFire.put(keyWithTimers.getKey(), timerData);
+          timerIt.remove();
+        }
+      }
+
+      if (keyWithTimers.getValue().isEmpty()) {
+        it.remove();
+      }
+    }
+    return toFire;
+  }
+
+  private void processElement(WindowedValue<KV<K, V>> windowedValue) throws Exception {
+    final KV<K, V> kv = windowedValue.getValue();
+    final WindowedValue<V> updatedWindowedValue = WindowedValue.of(kv.getValue(),
+        windowedValue.getTimestamp(),
+        windowedValue.getWindows(),
+        windowedValue.getPane());
+
+    KeyedWorkItem<K, V> kwi = KeyedWorkItems.elementsWorkItem(
+            kv.getKey(),
+            Collections.singletonList(updatedWindowedValue));
+
+    context.setElement(kwi, getStateInternalsForKey(kwi.key()));
+    fn.processElement(context);
+  }
+
+  private StateInternals<K> getStateInternalsForKey(K key) {
+    final ByteBuffer keyBytes;
+    try {
+      keyBytes = ByteBuffer.wrap(CoderUtils.encodeToByteArray(keyCoder, key));
+    } catch (CoderException e) {
+      throw new RuntimeException(e);
+    }
+    StateInternals<K> stateInternals = perKeyStateInternals.get(keyBytes);
+    if (stateInternals == null) {
+      stateInternals = stateInternalsFactory.stateInternalsForKey(key);
+      perKeyStateInternals.put(keyBytes, stateInternals);
+    }
+    return stateInternals;
+  }
+
+  private void registerActiveTimer(K key, TimerInternals.TimerData timer) {
+    final ByteBuffer keyBytes;
+    try {
+      keyBytes = ByteBuffer.wrap(CoderUtils.encodeToByteArray(keyCoder, key));
+    } catch (CoderException e) {
+      throw new RuntimeException(e);
+    }
+    Set<TimerInternals.TimerData> timersForKey = activeTimers.get(keyBytes);
+    if (timersForKey == null) {
+      timersForKey = new HashSet<>();
+    }
+    timersForKey.add(timer);
+    activeTimers.put(keyBytes, timersForKey);
+  }
+
+  private void unregisterActiveTimer(K key, TimerInternals.TimerData timer) {
+    final ByteBuffer keyBytes;
+    try {
+      keyBytes = ByteBuffer.wrap(CoderUtils.encodeToByteArray(keyCoder, key));
+    } catch (CoderException e) {
+      throw new RuntimeException(e);
+    }
+    Set<TimerInternals.TimerData> timersForKey = activeTimers.get(keyBytes);
+    if (timersForKey != null) {
+      timersForKey.remove(timer);
+      if (timersForKey.isEmpty()) {
+        activeTimers.remove(keyBytes);
+      } else {
+        activeTimers.put(keyBytes, timersForKey);
+      }
+    }
+  }
+
+  private void processWatermark(ApexStreamTuple.WatermarkTuple<?> mark) throws Exception {
+    this.inputWatermark = new Instant(mark.getTimestamp());
+    Multimap<ByteBuffer, TimerInternals.TimerData> timers = getTimersReadyToProcess(
+        mark.getTimestamp());
+    if (!timers.isEmpty()) {
+      for (ByteBuffer keyBytes : timers.keySet()) {
+        K key = CoderUtils.decodeFromByteArray(keyCoder, keyBytes.array());
+        KeyedWorkItem<K, V> kwi = KeyedWorkItems.<K, V>timersWorkItem(key, timers.get(keyBytes));
+        context.setElement(kwi, getStateInternalsForKey(kwi.key()));
+        fn.processElement(context);
+      }
+    }
+  }
+
+  private class ProcessContext extends GroupAlsoByWindowViaWindowSetDoFn<K, V, Iterable<V>, ?,
+      KeyedWorkItem<K, V>>.ProcessContext {
+
+    private final ApexTimerInternals timerInternals;
+    private StateInternals<K> stateInternals;
+    private KeyedWorkItem<K, V> element;
+
+    public ProcessContext(OldDoFn<KeyedWorkItem<K, V>, KV<K, Iterable<V>>> function,
+                          ApexTimerInternals timerInternals) {
+      function.super();
+      this.timerInternals = checkNotNull(timerInternals);
+    }
+
+    public void setElement(KeyedWorkItem<K, V> element, StateInternals<K> stateForKey) {
+      this.element = element;
+      this.stateInternals = stateForKey;
+    }
+
+    @Override
+    public KeyedWorkItem<K, V> element() {
+      return this.element;
+    }
+
+    @Override
+    public Instant timestamp() {
+      throw new UnsupportedOperationException(
+          "timestamp() is not available when processing KeyedWorkItems.");
+    }
+
+    @Override
+    public PipelineOptions getPipelineOptions() {
+      return serializedOptions.get();
+    }
+
+    @Override
+    public void output(KV<K, Iterable<V>> output) {
+      throw new UnsupportedOperationException(
+          "output() is not available when processing KeyedWorkItems.");
+    }
+
+    @Override
+    public void outputWithTimestamp(KV<K, Iterable<V>> output, Instant timestamp) {
+      throw new UnsupportedOperationException(
+          "outputWithTimestamp() is not available when processing KeyedWorkItems.");
+    }
+
+    @Override
+    public PaneInfo pane() {
+      throw new UnsupportedOperationException(
+          "pane() is not available when processing KeyedWorkItems.");
+    }
+
+    @Override
+    public BoundedWindow window() {
+      throw new UnsupportedOperationException(
+          "window() is not available when processing KeyedWorkItems.");
+    }
+
+    @Override
+    public WindowingInternals<KeyedWorkItem<K, V>, KV<K, Iterable<V>>> windowingInternals() {
+      return new WindowingInternals<KeyedWorkItem<K, V>, KV<K, Iterable<V>>>() {
+
+        @Override
+        public StateInternals<K> stateInternals() {
+          return stateInternals;
+        }
+
+        @Override
+        public void outputWindowedValue(KV<K, Iterable<V>> output, Instant timestamp,
+            Collection<? extends BoundedWindow> windows, PaneInfo pane) {
+          if (traceTuples) {
+            LOG.debug("\nemitting {} timestamp {}\n", output, timestamp);
+          }
+          ApexGroupByKeyOperator.this.output.emit(ApexStreamTuple.DataTuple.of(
+              WindowedValue.of(output, timestamp, windows, pane)));
+        }
+
+        @Override
+        public TimerInternals timerInternals() {
+          return timerInternals;
+        }
+
+        @Override
+        public Collection<? extends BoundedWindow> windows() {
+          throw new UnsupportedOperationException("windows() is not available in Streaming mode.");
+        }
+
+        @Override
+        public PaneInfo pane() {
+          throw new UnsupportedOperationException("pane() is not available in Streaming mode.");
+        }
+
+        @Override
+        public <T> void writePCollectionViewData(TupleTag<?> tag, Iterable<WindowedValue<T>> data,
+            Coder<T> elemCoder) throws IOException {
+          throw new RuntimeException("writePCollectionViewData() not available in Streaming mode.");
+        }
+
+        @Override
+        public <T> T sideInput(PCollectionView<T> view, BoundedWindow mainInputWindow) {
+          throw new RuntimeException("sideInput() is not available in Streaming mode.");
+        }
+      };
+    }
+
+    @Override
+    public <T> T sideInput(PCollectionView<T> view) {
+      throw new RuntimeException("sideInput() is not supported in Streaming mode.");
+    }
+
+    @Override
+    public <T> void sideOutput(TupleTag<T> tag, T output) {
+      // ignore the side output, this can happen when a user does not register
+      // side outputs but then outputs using a freshly created TupleTag.
+      throw new RuntimeException("sideOutput() is not available when grouping by window.");
+    }
+
+    @Override
+    public <T> void sideOutputWithTimestamp(TupleTag<T> tag, T output, Instant timestamp) {
+      sideOutput(tag, output);
+    }
+
+    @Override
+    protected <AggInputT, AggOutputT> Aggregator<AggInputT, AggOutputT> createAggregatorInternal(
+        String name, Combine.CombineFn<AggInputT, ?, AggOutputT> combiner) {
+      throw new UnsupportedOperationException();
+    }
+  }
+
+  /**
+   * An implementation of Beam's {@link TimerInternals}.
+   *
+   */
+  public class ApexTimerInternals implements TimerInternals {
+
+    @Override
+    public void setTimer(TimerData timerKey) {
+      registerActiveTimer(context.element().key(), timerKey);
+    }
+
+    @Override
+    public void deleteTimer(TimerData timerKey) {
+      unregisterActiveTimer(context.element().key(), timerKey);
+    }
+
+    @Override
+    public Instant currentProcessingTime() {
+      return Instant.now();
+    }
+
+    @Override
+    public Instant currentSynchronizedProcessingTime() {
+      // TODO Auto-generated method stub
+      return null;
+    }
+
+    @Override
+    public Instant currentInputWatermarkTime() {
+      return inputWatermark;
+    }
+
+    @Override
+    public Instant currentOutputWatermarkTime() {
+      // TODO Auto-generated method stub
+      return null;
+    }
+
+    @Override
+    public void setTimer(StateNamespace namespace, String timerId, Instant target,
+        TimeDomain timeDomain) {
+      throw new UnsupportedOperationException("Setting timer by ID not yet supported.");
+    }
+
+    @Override
+    public void deleteTimer(StateNamespace namespace, String timerId) {
+      throw new UnsupportedOperationException("Canceling of timer by ID is not yet supported.");
+    }
+
+  }
+
+  private class GroupByKeyStateInternalsFactory implements StateInternalsFactory<K>, Serializable {
+    private static final long serialVersionUID = 1L;
+
+    @Override
+    public StateInternals<K> stateInternalsForKey(K key) {
+      return getStateInternalsForKey(key);
+    }
+  }
+}


Mime
View raw message