You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by mw...@apache.org on 2020/08/27 09:25:53 UTC
[beam] branch master updated: [BEAM-10675] Clean up Python Load
Tests
This is an automated email from the ASF dual-hosted git repository.
mwalenia pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new fe2c822 [BEAM-10675] Clean up Python Load Tests
new d5201cf Merge pull request #12687 from kamilwu/load-tests-cleanup
fe2c822 is described below
commit fe2c822f82d58c49f845e7f8b303b8a61baa78da
Author: Kamil Wasilewski <ka...@polidea.com>
AuthorDate: Wed Aug 26 13:32:22 2020 +0200
[BEAM-10675] Clean up Python Load Tests
BEAM-10675 has been resolved, which enables Python GroupByKey streaming loads tests using SyntheticSource as a source of input. As a result, the old GroupByKey pipeline that read input from PubSub became obsolete and can be removed.
---
.../testing/load_tests/streaming/__init__.py | 16 ---
.../streaming/group_by_key_streaming_pipeline.py | 104 ------------------
.../streaming/group_by_key_streaming_test.py | 119 ---------------------
.../testing/load_tests/streaming/requirements.txt | 18 ----
4 files changed, 257 deletions(-)
diff --git a/sdks/python/apache_beam/testing/load_tests/streaming/__init__.py b/sdks/python/apache_beam/testing/load_tests/streaming/__init__.py
deleted file mode 100644
index cce3aca..0000000
--- a/sdks/python/apache_beam/testing/load_tests/streaming/__init__.py
+++ /dev/null
@@ -1,16 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
diff --git a/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_pipeline.py b/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_pipeline.py
deleted file mode 100644
index 03aa720..0000000
--- a/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_pipeline.py
+++ /dev/null
@@ -1,104 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-"""
-Streaming pipeline which reads, GBK and ungroup data from PubSub.
-Pipeline is cancelled when matcher achieves expected elements
-amount or reaches timeout. To cancel pipeline by matcher it is
-required to use TestRunner (TestDataflowRunner or TestDirectRunner).
-Data from pubsub is parsed into pair of strings so it
-would be possible group it by key.
-
-Values have to be reparsed again to bytes
-"""
-
-# pytype: skip-file
-
-from __future__ import absolute_import
-
-import argparse
-import logging
-
-import apache_beam as beam
-from apache_beam.io.gcp.pubsub import ReadFromPubSub
-from apache_beam.options.pipeline_options import PipelineOptions
-from apache_beam.options.pipeline_options import SetupOptions
-from apache_beam.options.pipeline_options import StandardOptions
-from apache_beam.testing.load_tests.load_test_metrics_utils import CountMessages
-from apache_beam.testing.load_tests.load_test_metrics_utils import MeasureTime
-from apache_beam.transforms import window
-
-
-def run(argv=None):
- class MessageParser(beam.DoFn):
- # It is required to parse messages for GBK operation.
- # Otherwise there are encoding problems.
- def process(self, item):
- if item.attributes:
- k, v = item.attributes.popitem()
- yield (str(k), str(v))
-
- class ParserToBytes(beam.DoFn):
- # Parsing to bytes is required for saving in PubSub.
- def process(self, item):
- _, v = item
- yield bytes(v, encoding='utf8')
-
- parser = argparse.ArgumentParser()
- parser.add_argument(
- '--output_topic',
- required=True,
- help=(
- 'Output PubSub topic of the form '
- '"projects/<PROJECT>/topic/<TOPIC>".'))
- parser.add_argument(
- '--input_subscription',
- help=(
- 'Input PubSub subscription of the form '
- '"projects/<PROJECT>/subscriptions/<SUBSCRIPTION>."'))
- parser.add_argument(
- '--metrics_namespace', help=('Namespace of metrics '
- '"string".'))
- known_args, pipeline_args = parser.parse_known_args(argv)
-
- pipeline_options = PipelineOptions(pipeline_args)
- pipeline_options.view_as(SetupOptions).save_main_session = True
- pipeline_options.view_as(StandardOptions).streaming = True
- p = beam.Pipeline(options=pipeline_options)
-
- # pylint: disable=expression-not-assigned
- (
- p
- | ReadFromPubSub(
- subscription=known_args.input_subscription, with_attributes=True)
- | 'Window' >> beam.WindowInto(window.FixedWindows(1000, 0))
- | 'Measure time: Start' >> beam.ParDo(
- MeasureTime(known_args.metrics_namespace))
- | 'Count messages' >> beam.ParDo(
- CountMessages(known_args.metrics_namespace))
- | 'Parse' >> beam.ParDo(MessageParser())
- | 'GroupByKey' >> beam.GroupByKey()
- | 'Ungroup' >> beam.FlatMap(lambda elm: [(elm[0], v) for v in elm[1]])
- | 'Measure time: End' >> beam.ParDo(
- MeasureTime(known_args.metrics_namespace))
- | 'Parse to bytes' >> beam.ParDo(ParserToBytes())
- | 'Write' >> beam.io.WriteToPubSub(topic=known_args.output_topic))
-
- result = p.run()
- result.wait_until_finish()
- logging.error(result)
- return result
diff --git a/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_test.py b/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_test.py
deleted file mode 100644
index 75440f8..0000000
--- a/sdks/python/apache_beam/testing/load_tests/streaming/group_by_key_streaming_test.py
+++ /dev/null
@@ -1,119 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License"); you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-"""
-Performance GBK streaming test that uses PubSub SyntheticSources
-messages.
-
-Test requires --test-pipeline-options with following options:
-* --pubsub_topic_name=name - name of PubSub topic which is
- already created
-* --project=project-name
-* --num_of_records=1000 - expected number of records
-* --runner=TestDataflowRunner or TestDirectRunner - only
- test runners supports matchers
-
-Optional pipeline options:
-* --timeout=1000 max time that test will be run and wait
- for all messages.
-* --publish_to_big_query=true/false
-* --metrics_dataset=python_load_tests
-* --metrics_table=gbk_stream
-"""
-
-# pytype: skip-file
-
-from __future__ import absolute_import
-
-import logging
-import uuid
-
-from hamcrest.core.core.allof import all_of
-
-from apache_beam.io.gcp.tests.pubsub_matcher import PubSubMessageMatcher
-from apache_beam.testing import test_utils
-from apache_beam.testing.load_tests.load_test import LoadTest
-from apache_beam.testing.load_tests.streaming import group_by_key_streaming_pipeline
-
-DEFAULT_TIMEOUT = 800
-WAIT_UNTIL_FINISH_DURATION = 1 * 60 * 1000
-
-
-class GroupByKeyStreamingTest(LoadTest):
- ID_LABEL = 'id'
-
- def __init__(self):
- super(GroupByKeyStreamingTest, self).__init__()
- self.topic_short_name = self.pipeline.get_option('pubsub_topic_name')
- self.setup_pubsub()
-
- timeout = self.pipeline.get_option('timeout') or DEFAULT_TIMEOUT
- expected_num_of_records = self.pipeline.get_option('num_of_records')
- pubsub_msg_verifier = PubSubMessageMatcher(
- self.project_id,
- self.output_sub.name,
- expected_msg_len=int(expected_num_of_records),
- timeout=int(timeout))
-
- self.extra_opts = {
- 'input_subscription': self.input_sub.name,
- 'output_topic': self.output_topic.name,
- 'metrics_namespace': self.metrics_namespace,
- 'wait_until_finish_duration': WAIT_UNTIL_FINISH_DURATION,
- 'on_success_matcher': all_of(pubsub_msg_verifier)
- }
-
- def setup_pubsub(self):
- self.uuid = str(uuid.uuid4())
-
- # Set up PubSub environment.
- from google.cloud import pubsub
- self.pub_client = pubsub.PublisherClient()
- input_topic_full_name = "projects/{}/topics/{}" \
- .format(self.project_id, self.topic_short_name)
- self.input_topic = self.pub_client.get_topic(input_topic_full_name)
- self.output_topic_name = self.topic_short_name + '_out_' + self.uuid
- self.output_topic = self.pub_client.create_topic(
- self.pub_client.topic_path(self.project_id, self.output_topic_name))
-
- self.sub_client = pubsub.SubscriberClient()
- self.input_sub_name = self.topic_short_name + '_sub_in_' + self.uuid
- self.input_sub = self.sub_client.create_subscription(
- self.sub_client.subscription_path(self.project_id, self.input_sub_name),
- self.input_topic.name)
- self.output_sub_name = self.topic_short_name + '_sub_out_' + self.uuid
- self.output_sub = self.sub_client.create_subscription(
- self.sub_client.subscription_path(
- self.project_id, self.output_sub_name),
- self.output_topic.name,
- ack_deadline_seconds=60)
-
- def test(self):
- args = self.pipeline.get_full_options_as_args(**self.extra_opts)
- self.result = group_by_key_streaming_pipeline.run(args)
-
- # If the test timeouts on Dataflow it may not cleanup pubsub after
- def cleanup(self):
- test_utils.cleanup_subscriptions(
- self.sub_client, [self.input_sub, self.output_sub])
- test_utils.cleanup_topics(
- self.pub_client, [self.input_topic, self.output_topic])
-
-
-if __name__ == '__main__':
- logging.basicConfig(level=logging.INFO)
- GroupByKeyStreamingTest().run()
diff --git a/sdks/python/apache_beam/testing/load_tests/streaming/requirements.txt b/sdks/python/apache_beam/testing/load_tests/streaming/requirements.txt
deleted file mode 100644
index a8e68b8..0000000
--- a/sdks/python/apache_beam/testing/load_tests/streaming/requirements.txt
+++ /dev/null
@@ -1,18 +0,0 @@
-#
-# Licensed to the Apache Software Foundation (ASF) under one or more
-# contributor license agreements. See the NOTICE file distributed with
-# this work for additional information regarding copyright ownership.
-# The ASF licenses this file to You under the Apache License, Version 2.0
-# (the "License")# you may not use this file except in compliance with
-# the License. You may obtain a copy of the License at
-#
-# http://www.apache.org/licenses/LICENSE-2.0
-#
-# Unless required by applicable law or agreed to in writing, software
-# distributed under the License is distributed on an "AS IS" BASIS,
-# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-# See the License for the specific language governing permissions and
-# limitations under the License.
-#
-
-pyhamcrest
\ No newline at end of file