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 2017/05/24 22:15:20 UTC
[1/2] beam git commit: Adding a snippet for metrics
Repository: beam
Updated Branches:
refs/heads/master 4084f71a1 -> c0d19f9f7
Adding a snippet for metrics
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/f32cacb7
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/f32cacb7
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/f32cacb7
Branch: refs/heads/master
Commit: f32cacb7313b8352ddc054cd1b88d7a0462550db
Parents: 4084f71
Author: Pablo <pa...@google.com>
Authored: Mon May 22 15:39:44 2017 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed May 24 15:14:41 2017 -0700
----------------------------------------------------------------------
.../examples/snippets/snippets_test.py | 62 ++++++++++++++++++++
1 file changed, 62 insertions(+)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/beam/blob/f32cacb7/sdks/python/apache_beam/examples/snippets/snippets_test.py
----------------------------------------------------------------------
diff --git a/sdks/python/apache_beam/examples/snippets/snippets_test.py b/sdks/python/apache_beam/examples/snippets/snippets_test.py
index 6654fef..e302465 100644
--- a/sdks/python/apache_beam/examples/snippets/snippets_test.py
+++ b/sdks/python/apache_beam/examples/snippets/snippets_test.py
@@ -32,6 +32,8 @@ from apache_beam import typehints
from apache_beam.coders.coders import ToStringCoder
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.examples.snippets import snippets
+from apache_beam.metrics import Metrics
+from apache_beam.metrics.metric import MetricsFilter
from apache_beam.testing.util import assert_that
from apache_beam.testing.util import equal_to
from apache_beam.utils.windowed_value import WindowedValue
@@ -689,6 +691,66 @@ class SnippetsTest(unittest.TestCase):
expect = ['a; a@example.com; x4312', 'b; b@example.com; x8452']
self.assertEqual(expect, self.get_output(result_path))
+ def test_model_use_and_query_metrics(self):
+ """DebuggingWordCount example snippets."""
+
+ import re
+
+ p = TestPipeline() # Use TestPipeline for testing.
+ words = p | beam.Create(['albert', 'sam', 'mark', 'sarah',
+ 'swati', 'daniel', 'andrea'])
+
+ # pylint: disable=unused-variable
+ # [START metrics_usage_example]
+ class FilterTextFn(beam.DoFn):
+ """A DoFn that filters for a specific key based on a regex."""
+
+ def __init__(self, pattern):
+ self.pattern = pattern
+ # A custom metric can track values in your pipeline as it runs. Create
+ # custom metrics to count unmatched words, and know the distribution of
+ # word lengths in the input PCollection.
+ self.word_len_dist = Metrics.distribution(self.__class__,
+ 'word_len_dist')
+ self.unmatched_words = Metrics.counter(self.__class__,
+ 'unmatched_words')
+
+ def process(self, element):
+ word = element
+ self.word_len_dist.update(len(word))
+ if re.match(self.pattern, word):
+ yield element
+ else:
+ self.unmatched_words.inc()
+
+ filtered_words = (
+ words | 'FilterText' >> beam.ParDo(FilterTextFn('s.*')))
+ # [END metrics_usage_example]
+ # pylint: enable=unused-variable
+
+ # [START metrics_check_values_example]
+ result = p.run()
+ result.wait_until_finish()
+
+ custom_distribution = result.metrics().query(
+ MetricsFilter().with_name('word_len_dist'))['distributions']
+ custom_counter = result.metrics().query(
+ MetricsFilter().with_name('unmatched_words'))['counters']
+
+ if custom_distribution:
+ logging.info('The average word length was %d',
+ custom_distribution[0].committed.mean)
+ if custom_counter:
+ logging.info('There were %d words that did not match the filter.',
+ custom_counter[0].committed)
+ # [END metrics_check_values_example]
+
+ # There should be 4 words that did not match
+ self.assertEqual(custom_counter[0].committed, 4)
+ # The shortest word is 3 characters, the longest is 6
+ self.assertEqual(custom_distribution[0].committed.min, 3)
+ self.assertEqual(custom_distribution[0].committed.max, 6)
+
def test_model_join_using_side_inputs(self):
name_list = ['a', 'b']
email_list = [['a', 'a@example.com'], ['b', 'b@example.com']]
[2/2] beam git commit: Closes #3200
Posted by ro...@apache.org.
Closes #3200
Project: http://git-wip-us.apache.org/repos/asf/beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/beam/commit/c0d19f9f
Tree: http://git-wip-us.apache.org/repos/asf/beam/tree/c0d19f9f
Diff: http://git-wip-us.apache.org/repos/asf/beam/diff/c0d19f9f
Branch: refs/heads/master
Commit: c0d19f9f7cccb222819dba26be52d85262f3f18f
Parents: 4084f71 f32cacb
Author: Robert Bradshaw <ro...@google.com>
Authored: Wed May 24 15:14:48 2017 -0700
Committer: Robert Bradshaw <ro...@google.com>
Committed: Wed May 24 15:14:48 2017 -0700
----------------------------------------------------------------------
.../examples/snippets/snippets_test.py | 62 ++++++++++++++++++++
1 file changed, 62 insertions(+)
----------------------------------------------------------------------