You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by ro...@apache.org on 2016/07/11 23:48:08 UTC
[1/2] incubator-beam git commit: Internal cleanup.
Repository: incubator-beam
Updated Branches:
refs/heads/python-sdk 73168b220 -> b5daba37f
Internal cleanup.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c7b8c688
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c7b8c688
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c7b8c688
Branch: refs/heads/python-sdk
Commit: c7b8c688e1bf3d44f6c9072d58a173a2618a5e13
Parents: 73168b2
Author: Robert Bradshaw <ro...@google.com>
Authored: Mon Jul 11 15:20:25 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 11 15:20:25 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/coders/coder_impl.pxd | 3 ++-
sdks/python/apache_beam/pvalue.py | 3 +++
.../apache_beam/runners/dataflow_runner.py | 25 +++++++++++---------
sdks/python/apache_beam/transforms/core.py | 18 ++++++++++----
4 files changed, 33 insertions(+), 16 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c7b8c688/sdks/python/apache_beam/coders/coder_impl.pxd
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/coders/coder_impl.pxd b/sdks/python/apache_beam/coders/coder_impl.pxd
index a86d08b..0c92c5b 100644
--- a/sdks/python/apache_beam/coders/coder_impl.pxd
+++ b/sdks/python/apache_beam/coders/coder_impl.pxd
@@ -29,7 +29,8 @@ from .stream cimport InputStream, OutputStream
cdef object loads, dumps, create_InputStream, create_OutputStream
-cdef type WindowedValue
+# Temporarily untyped to allow monkeypatching on failed import.
+#cdef type WindowedValue
cdef class CoderImpl(object):
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c7b8c688/sdks/python/apache_beam/pvalue.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/pvalue.py b/sdks/python/apache_beam/pvalue.py
index 78ff209..8552a45 100644
--- a/sdks/python/apache_beam/pvalue.py
+++ b/sdks/python/apache_beam/pvalue.py
@@ -197,6 +197,9 @@ class DoOutputsTuple(object):
# Transfer the producer from the DoOutputsTuple to the resulting
# PCollection.
pcoll.producer = self.producer
+ # Add this as an output to both the inner ParDo and the outer _MultiParDo
+ # PTransforms.
+ self.producer.parts[0].add_output(pcoll, tag)
self.producer.add_output(pcoll, tag)
self._pcolls[tag] = pcoll
return pcoll
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c7b8c688/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 43a50ba..24edb05 100644
--- a/sdks/python/apache_beam/runners/dataflow_runner.py
+++ b/sdks/python/apache_beam/runners/dataflow_runner.py
@@ -190,7 +190,8 @@ class DataflowPipelineRunner(PipelineRunner):
return self._get_cloud_encoding(self._get_coder(typehint,
window_coder=window_coder))
- def _get_coder(self, typehint, window_coder):
+ @staticmethod
+ def _get_coder(typehint, window_coder):
"""Returns a coder based on a typehint object."""
if window_coder:
return coders.WindowedValueCoder(
@@ -360,27 +361,20 @@ class DataflowPipelineRunner(PipelineRunner):
# Attach side inputs.
si_dict = {}
- si_tags_and_types = []
+ lookup_label = lambda side_pval: self._cache.get_pvalue(side_pval).step_name
for side_pval in transform_node.side_inputs:
assert isinstance(side_pval, PCollectionView)
- side_input_step = self._cache.get_pvalue(side_pval)
- si_label = side_input_step.step_name
+ si_label = lookup_label(side_pval)
si_dict[si_label] = {
'@type': 'OutputReference',
PropertyNames.STEP_NAME: si_label,
PropertyNames.OUTPUT_NAME: PropertyNames.OUT}
- # The label for the side input step will appear as a 'tag' property for
- # the side input source specification. Its type (singleton or iterator)
- # will also be used to read the entire source or just first element.
- si_tags_and_types.append((si_label, side_pval.__class__,
- side_pval._view_options())) # pylint: disable=protected-access
# Now create the step for the ParDo transform being handled.
step = self._add_step(
TransformNames.DO, transform_node.full_label, transform_node,
transform_node.transform.side_output_tags)
- fn_data = (transform.fn, transform.args, transform.kwargs,
- si_tags_and_types, transform_node.inputs[0].windowing)
+ fn_data = self._pardo_fn_data(transform_node, lookup_label)
step.add_property(PropertyNames.SERIALIZED_FN, pickler.dumps(fn_data))
step.add_property(
PropertyNames.PARALLEL_INPUT,
@@ -415,6 +409,15 @@ class DataflowPipelineRunner(PipelineRunner):
'%s_%s' % (PropertyNames.OUT, side_tag))})
step.add_property(PropertyNames.OUTPUT_INFO, outputs)
+ @staticmethod
+ def _pardo_fn_data(transform_node, get_label):
+ transform = transform_node.transform
+ si_tags_and_types = [ # pylint: disable=protected-access
+ (get_label(side_pval), side_pval.__class__, side_pval._view_options())
+ for side_pval in transform_node.side_inputs]
+ return (transform.fn, transform.args, transform.kwargs, si_tags_and_types,
+ transform_node.inputs[0].windowing)
+
def apply_CombineValues(self, transform, pcoll):
return pvalue.PCollection(pcoll.pipeline)
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c7b8c688/sdks/python/apache_beam/transforms/core.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/transforms/core.py b/sdks/python/apache_beam/transforms/core.py
index 9b477db..15247f2 100644
--- a/sdks/python/apache_beam/transforms/core.py
+++ b/sdks/python/apache_beam/transforms/core.py
@@ -28,7 +28,6 @@ from apache_beam.internal import util
from apache_beam.transforms import ptransform
from apache_beam.transforms import window
from apache_beam.transforms.ptransform import PTransform
-from apache_beam.transforms.ptransform import ptransform_fn
from apache_beam.transforms.ptransform import PTransformWithSideInputs
from apache_beam.transforms.window import MIN_TIMESTAMP
from apache_beam.transforms.window import OutputTimeFn
@@ -856,8 +855,7 @@ class CombineGlobally(PTransform):
| typed(Map('InjectDefault', lambda _, s: s, view)))
-@ptransform_fn
-def CombinePerKey(pcoll, fn, *args, **kwargs): # pylint: disable=invalid-name
+class CombinePerKey(PTransformWithSideInputs):
"""A per-key Combine transform.
Identifies sets of values associated with the same key in the input
@@ -876,7 +874,19 @@ def CombinePerKey(pcoll, fn, *args, **kwargs): # pylint: disable=invalid-name
Returns:
A PObject holding the result of the combine operation.
"""
- return pcoll | GroupByKey() | CombineValues('Combine', fn, *args, **kwargs)
+
+ def make_fn(self, fn):
+ self._fn_label = ptransform.label_from_callable(fn)
+ return fn if isinstance(fn, CombineFn) else CombineFn.from_callable(fn)
+
+ def default_label(self):
+ return '%s(%s)' % (self.__class__.__name__, self._fn_label)
+
+ def apply(self, pcoll):
+ args, kwargs = util.insert_values_in_args(
+ self.args, self.kwargs, self.side_inputs)
+ return pcoll | GroupByKey() | CombineValues('Combine',
+ self.fn, *args, **kwargs)
# TODO(robertwb): Rename to CombineGroupedValues?
[2/2] incubator-beam git commit: Closes #633
Posted by ro...@apache.org.
Closes #633
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/b5daba37
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/b5daba37
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/b5daba37
Branch: refs/heads/python-sdk
Commit: b5daba37fa285d428a1d8ffe9106b7fe265b2914
Parents: 73168b2 c7b8c68
Author: Robert Bradshaw <ro...@google.com>
Authored: Mon Jul 11 16:47:24 2016 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Mon Jul 11 16:47:24 2016 -0700
----------------------------------------------------------------------
sdks/python/apache_beam/coders/coder_impl.pxd | 3 ++-
sdks/python/apache_beam/pvalue.py | 3 +++
.../apache_beam/runners/dataflow_runner.py | 25 +++++++++++---------
sdks/python/apache_beam/transforms/core.py | 18 ++++++++++----
4 files changed, 33 insertions(+), 16 deletions(-)
----------------------------------------------------------------------