Make StreamingPCollectionViewWriterFn and its data public This enables a smooth migration path whereby the runner can notice the class and extract these values, to run them without OldDoFn.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d70a3a47 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d70a3a47 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d70a3a47 Branch: refs/heads/gearpump-runner Commit: d70a3a47ee39fab8c4514a7bda4126827baf65a1 Parents: 7da1a3d Author: Kenneth Knowles <[email protected]> Authored: Tue Aug 9 14:36:28 2016 -0700 Committer: Kenneth Knowles <[email protected]> Committed: Tue Aug 9 14:36:28 2016 -0700 ---------------------------------------------------------------------- .../org/apache/beam/runners/dataflow/DataflowRunner.java | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d70a3a47/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java ---------------------------------------------------------------------- diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java index fadd9c7..6da33fa 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java @@ -2375,7 +2375,8 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> { * A specialized {@link OldDoFn} for writing the contents of a {@link PCollection} * to a streaming {@link PCollectionView} backend implementation. */ - private static class StreamingPCollectionViewWriterFn<T> + @Deprecated + public static class StreamingPCollectionViewWriterFn<T> extends OldDoFn<Iterable<T>, T> implements OldDoFn.RequiresWindowAccess { private final PCollectionView<?> view; private final Coder<T> dataCoder; @@ -2390,6 +2391,14 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> { this.dataCoder = dataCoder; } + public PCollectionView<?> getView() { + return view; + } + + public Coder<T> getDataCoder() { + return dataCoder; + } + @Override public void processElement(ProcessContext c) throws Exception { List<WindowedValue<T>> output = new ArrayList<>();
