beam-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From dhalp...@apache.org
Subject [1/2] incubator-beam git commit: Make step encodings consistently use WindowedValueCoders
Date Sat, 16 Jul 2016 00:11:22 GMT
Repository: incubator-beam
Updated Branches:
  refs/heads/python-sdk 1305c108a -> b8e3a7b52


Make step encodings consistently use WindowedValueCoders

This change is needed for size-estimation aggregators.


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

Branch: refs/heads/python-sdk
Commit: 3d8434ec95327d4bb374d38196e883a49f2a76d4
Parents: 1305c10
Author: Charles Chen <ccy@google.com>
Authored: Fri Jul 15 14:37:43 2016 -0700
Committer: Charles Chen <ccy@google.com>
Committed: Fri Jul 15 14:37:43 2016 -0700

----------------------------------------------------------------------
 sdks/python/apache_beam/runners/dataflow_runner.py | 12 +++++++++++-
 sdks/python/apache_beam/utils/dependency.py        |  6 +++++-
 2 files changed, 16 insertions(+), 2 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3d8434ec/sdks/python/apache_beam/runners/dataflow_runner.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/dataflow_runner.py b/sdks/python/apache_beam/runners/dataflow_runner.py
index 45bfb6e..4fafc9f 100644
--- a/sdks/python/apache_beam/runners/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow_runner.py
@@ -530,6 +530,12 @@ class DataflowPipelineRunner(PipelineRunner):
     else:
       coder = transform.source.coder
 
+    # Wrap coder in WindowedValueCoder: this is necessary as the encoding of a
+    # step should be the type of value outputted by each step.  Read steps
+    # automatically wrap output values in a WindowedValue wrapper, if necessary.
+    # This is also necessary for proper encoding for size estimation.
+    coder = coders.WindowedValueCoder(coder)
+
     step.encoding = self._get_cloud_encoding(coder)
     step.add_property(
         PropertyNames.OUTPUT_INFO,
@@ -595,7 +601,11 @@ class DataflowPipelineRunner(PipelineRunner):
           'Sink %r has unexpected format %s.' % (
               transform.sink, transform.sink.format))
     step.add_property(PropertyNames.FORMAT, transform.sink.format)
-    step.encoding = self._get_cloud_encoding(transform.sink.coder)
+
+    # Wrap coder in WindowedValueCoder: this is necessary for proper encoding
+    # for size estimation.
+    coder = coders.WindowedValueCoder(transform.sink.coder)
+    step.encoding = self._get_cloud_encoding(coder)
     step.add_property(PropertyNames.ENCODING, step.encoding)
     step.add_property(
         PropertyNames.PARALLEL_INPUT,

http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/3d8434ec/sdks/python/apache_beam/utils/dependency.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/utils/dependency.py b/sdks/python/apache_beam/utils/dependency.py
index b809cf2..bd0983c 100644
--- a/sdks/python/apache_beam/utils/dependency.py
+++ b/sdks/python/apache_beam/utils/dependency.py
@@ -434,7 +434,11 @@ def get_required_container_version():
   except pkg.DistributionNotFound:
     # This case covers Apache Beam end-to-end testing scenarios. All these tests
     # will run with a special container version.
-    return 'beamhead'
+    #
+    # TODO(ccy): change this back to 'beamhead' when worker support for the
+    # recent change to use WindowedValueCoders in Sources and Sinks is rolled
+    # out.
+    return 'beamhead-02'
 
 
 def _download_pypi_sdk_package(temp_dir):


Mime
View raw message