You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2017/09/06 20:59:20 UTC
[1/2] beam git commit: Incorporate partial writeout fix
Repository: beam
Updated Branches:
refs/heads/master 1882939c0 -> 6dfd3a2b7
Incorporate partial writeout fix
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/095b09bb
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/095b09bb
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/095b09bb
Branch: refs/heads/master
Commit: 095b09bbe7de5da6479e0de639ea6a9d10a05079
Parents: 1882939
Author: Maria Garcia Herrero <ma...@google.com>
Authored: Tue Aug 29 15:26:49 2017 -0700
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Sep 6 13:59:08 2017 -0700
----------------------------------------------------------------------
.../runners/direct/evaluation_context.py | 28 ++++++++++++-----
.../apache_beam/runners/direct/executor.py | 2 +-
.../runners/direct/transform_evaluator.py | 32 ++++++++------------
sdks/python/apache_beam/runners/direct/util.py | 7 +++--
4 files changed, 40 insertions(+), 29 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/095b09bb/sdks/python/apache_beam/runners/direct/evaluation_context.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct/evaluation_context.py b/sdks/python/apache_beam/runners/direct/evaluation_context.py
index 9913813..a927b09 100644
--- a/sdks/python/apache_beam/runners/direct/evaluation_context.py
+++ b/sdks/python/apache_beam/runners/direct/evaluation_context.py
@@ -20,6 +20,7 @@
from __future__ import absolute_import
import collections
+import copy
import threading
from apache_beam.runners.direct.clock import Clock
@@ -241,6 +242,10 @@ class EvaluationContext(object):
counter.name, counter.combine_fn)
merged_counter.accumulator.merge([counter.accumulator])
+ # Commit partial GBK states
+ existing_keyed_state = self._transform_keyed_states[result.transform]
+ for k, v in result.partial_keyed_state.iteritems():
+ existing_keyed_state[k] = v
return committed_bundles
def get_aggregator_values(self, aggregator_or_name):
@@ -316,16 +321,25 @@ class DirectUnmergedState(InMemoryUnmergedState):
def __init__(self):
super(DirectUnmergedState, self).__init__(defensive_copy=False)
+ # TODO(mariagh): make a selective deepcopy of just what is needed
+ # to preserve the state while a bundle is processed.
+ def clone(self):
+ return copy.deepcopy(self)
+
class DirectStepContext(object):
"""Context for the currently-executing step."""
- def __init__(self, keyed_existing_state):
- self.keyed_existing_state = keyed_existing_state
+ def __init__(self, existing_keyed_state):
+ self.existing_keyed_state = existing_keyed_state
+ # In order to avoid partial writes of a bundle, every time
+ # existing_keyed_state is accessed, a copy of the state is made
+ # to be transferred to the bundle state once the bundle is committed.
+ self.partial_keyed_state = {}
def get_keyed_state(self, key):
- # TODO(ccy): consider implementing transactional copy on write semantics
- # for state so that work items can be safely retried.
- if not self.keyed_existing_state.get(key):
- self.keyed_existing_state[key] = DirectUnmergedState()
- return self.keyed_existing_state[key]
+ if not self.existing_keyed_state.get(key):
+ self.existing_keyed_state[key] = DirectUnmergedState()
+ if not self.partial_keyed_state.get(key):
+ self.partial_keyed_state[key] = self.existing_keyed_state[key].clone()
+ return self.partial_keyed_state[key]
http://git-wip-us.apache.org/repos/asf/beam/blob/095b09bb/sdks/python/apache_beam/runners/direct/executor.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct/executor.py b/sdks/python/apache_beam/runners/direct/executor.py
index 890aa88..51fe908 100644
--- a/sdks/python/apache_beam/runners/direct/executor.py
+++ b/sdks/python/apache_beam/runners/direct/executor.py
@@ -538,7 +538,7 @@ class _ExecutorServiceParallelExecutor(object):
update.unprocessed_bundle)
else:
assert update.exception
- logging.warning('A task failed with exception.\n %s',
+ logging.warning('A task failed with exception: %s',
update.exception)
self._executor.visible_updates.offer(
_ExecutorServiceParallelExecutor._VisibleExecutorUpdate(
http://git-wip-us.apache.org/repos/asf/beam/blob/095b09bb/sdks/python/apache_beam/runners/direct/transform_evaluator.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct/transform_evaluator.py b/sdks/python/apache_beam/runners/direct/transform_evaluator.py
index 8906ccb..26d2019 100644
--- a/sdks/python/apache_beam/runners/direct/transform_evaluator.py
+++ b/sdks/python/apache_beam/runners/direct/transform_evaluator.py
@@ -302,8 +302,7 @@ class _BoundedReadEvaluator(_TransformEvaluator):
with self._source.reader() as reader:
bundles = _read_values_to_bundles(reader)
- return TransformResult(
- self._applied_ptransform, bundles, [], None, None)
+ return TransformResult(self, bundles, [], None, None)
class _TestStreamEvaluator(_TransformEvaluator):
@@ -356,9 +355,9 @@ class _TestStreamEvaluator(_TransformEvaluator):
self.current_index + 1, timestamp=self.watermark))
unprocessed_bundles.append(unprocessed_bundle)
hold = self.watermark
+
return TransformResult(
- self._applied_ptransform, self.bundles, unprocessed_bundles, None,
- {None: hold})
+ self, self.bundles, unprocessed_bundles, None, {None: hold})
class _PubSubSubscriptionWrapper(object):
@@ -442,9 +441,9 @@ class _PubSubReadEvaluator(_TransformEvaluator):
input_pvalue = pvalue.PBegin(self._applied_ptransform.transform.pipeline)
unprocessed_bundle = self._evaluation_context.create_bundle(
input_pvalue)
- return TransformResult(
- self._applied_ptransform, bundles,
- [unprocessed_bundle], None, {None: Timestamp.of(time.time())})
+
+ return TransformResult(self, bundles, [unprocessed_bundle], None,
+ {None: Timestamp.of(time.time())})
class _FlattenEvaluator(_TransformEvaluator):
@@ -467,8 +466,7 @@ class _FlattenEvaluator(_TransformEvaluator):
def finish_bundle(self):
bundles = [self.bundle]
- return TransformResult(
- self._applied_ptransform, bundles, [], None, None)
+ return TransformResult(self, bundles, [], None, None)
class _TaggedReceivers(dict):
@@ -557,7 +555,7 @@ class _ParDoEvaluator(_TransformEvaluator):
bundles = self._tagged_receivers.values()
result_counters = self._counter_factory.get_counters()
return TransformResult(
- self._applied_ptransform, bundles, [], result_counters, None,
+ self, bundles, [], result_counters, None,
self._tagged_receivers.undeclared_in_memory_tag_values)
@@ -621,7 +619,7 @@ class _GroupByKeyOnlyEvaluator(_TransformEvaluator):
gbk_result = []
# TODO(ccy): perhaps we can clean this up to not use this
# internal attribute of the DirectStepContext.
- for encoded_k in self.step_context.keyed_existing_state:
+ for encoded_k in self.step_context.existing_keyed_state:
# Ignore global state.
if encoded_k is None:
continue
@@ -647,8 +645,7 @@ class _GroupByKeyOnlyEvaluator(_TransformEvaluator):
self.global_state.set_timer(
None, '', TimeDomain.WATERMARK, WatermarkManager.WATERMARK_POS_INF)
- return TransformResult(
- self._applied_ptransform, bundles, [], None, {None: hold})
+ return TransformResult(self, bundles, [], None, {None: hold})
class _StreamingGroupByKeyOnlyEvaluator(_TransformEvaluator):
@@ -701,8 +698,7 @@ class _StreamingGroupByKeyOnlyEvaluator(_TransformEvaluator):
kwi = KeyedWorkItem(encoded_k, elements=vs)
bundle.add(GlobalWindows.windowed_value(kwi))
- return TransformResult(
- self._applied_ptransform, bundles, [], None, None)
+ return TransformResult(self, bundles, [], None, None)
class _StreamingGroupAlsoByWindowEvaluator(_TransformEvaluator):
@@ -762,8 +758,7 @@ class _StreamingGroupAlsoByWindowEvaluator(_TransformEvaluator):
bundle.add(item)
bundles.append(bundle)
- return TransformResult(
- self._applied_ptransform, bundles, [], None, self.keyed_holds)
+ return TransformResult(self, bundles, [], None, self.keyed_holds)
class _NativeWriteEvaluator(_TransformEvaluator):
@@ -827,5 +822,4 @@ class _NativeWriteEvaluator(_TransformEvaluator):
self.global_state.set_timer(
None, '', TimeDomain.WATERMARK, WatermarkManager.WATERMARK_POS_INF)
- return TransformResult(
- self._applied_ptransform, [], [], None, {None: hold})
+ return TransformResult(self, [], [], None, {None: hold})
http://git-wip-us.apache.org/repos/asf/beam/blob/095b09bb/sdks/python/apache_beam/runners/direct/util.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/runners/direct/util.py b/sdks/python/apache_beam/runners/direct/util.py
index 10f7b29..96a6ee2 100644
--- a/sdks/python/apache_beam/runners/direct/util.py
+++ b/sdks/python/apache_beam/runners/direct/util.py
@@ -26,10 +26,10 @@ from __future__ import absolute_import
class TransformResult(object):
"""Result of evaluating an AppliedPTransform with a TransformEvaluator."""
- def __init__(self, applied_ptransform, uncommitted_output_bundles,
+ def __init__(self, transform_evaluator, uncommitted_output_bundles,
unprocessed_bundles, counters, keyed_watermark_holds,
undeclared_tag_values=None):
- self.transform = applied_ptransform
+ self.transform = transform_evaluator._applied_ptransform
self.uncommitted_output_bundles = uncommitted_output_bundles
self.unprocessed_bundles = unprocessed_bundles
self.counters = counters
@@ -47,6 +47,9 @@ class TransformResult(object):
# Populated by the TransformExecutor.
self.logical_metric_updates = None
+ step_context = transform_evaluator._execution_context.get_step_context()
+ self.partial_keyed_state = step_context.partial_keyed_state
+
class TimerFiring(object):
"""A single instance of a fired timer."""
[2/2] beam git commit: This closes #3787
Posted by al...@apache.org.
This closes #3787
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/6dfd3a2b
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/6dfd3a2b
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/6dfd3a2b
Branch: refs/heads/master
Commit: 6dfd3a2b7d748bff049a34ea39cb5e2a0c45dba0
Parents: 1882939 095b09b
Author: Ahmet Altay <al...@google.com>
Authored: Wed Sep 6 13:59:11 2017 -0700
Committer: Ahmet Altay <al...@google.com>
Committed: Wed Sep 6 13:59:11 2017 -0700
----------------------------------------------------------------------
.../runners/direct/evaluation_context.py | 28 ++++++++++++-----
.../apache_beam/runners/direct/executor.py | 2 +-
.../runners/direct/transform_evaluator.py | 32 ++++++++------------
sdks/python/apache_beam/runners/direct/util.py | 7 +++--
4 files changed, 40 insertions(+), 29 deletions(-)
----------------------------------------------------------------------