[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[
https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=177132=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-177132
]
ASF GitHub Bot logged work on BEAM-5993:
Author: ASF GitHub Bot
Created on: 19/Dec/18 18:51
Start Date: 19/Dec/18 18:51
Worklog Time Spent: 10m
Work Description: pabloem closed pull request #7020: BEAM-5993 Create
SideInput Load test
URL: https://github.com/apache/beam/pull/7020
This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:
As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):
diff --git a/sdks/python/apache_beam/testing/load_tests/sideinput_test.py
b/sdks/python/apache_beam/testing/load_tests/sideinput_test.py
new file mode 100644
index ..435f44c02eb4
--- /dev/null
+++ b/sdks/python/apache_beam/testing/load_tests/sideinput_test.py
@@ -0,0 +1,203 @@
+#
+# 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.
+#
+"""
+This is SideInput load test with Synthetic Source. Besides of the standard
+input options there are additional options:
+* number_of_counter_operations - number of pardo operations
+* project (optional) - the gcp project in case of saving
+metrics in Big Query (in case of Dataflow Runner
+it is required to specify project of runner),
+* metrics_table (optional) - name of BigQuery table where metrics
+will be stored,
+in case of lack of any of both options metrics won't be saved
+* metrics_dataset (optional) - name of BigQuery dataset where metrics
+will be stored,
+in case of lack of all three options metrics won't be saved
+* input_options - options for Synthetic Sources.
+
+To run test on DirectRunner
+
+python setup.py nosetests \
+--project=big-query-project
+--metrics_dataset=python_load_tests
+--metrics_table=side_input
+--test-pipeline-options="
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 300,
+\"key_size\": 5,
+\"value_size\":15,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+ " \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+To run test on other runner (ex. Dataflow):
+
+python setup.py nosetests \
+--test-pipeline-options="
+--runner=TestDataflowRunner
+--project=...
+--metrics_dataset=python_load_tests
+--metrics_table=side_input
+--staging_location=gs://...
+--temp_location=gs://...
+--sdk_location=./dist/apache-beam-x.x.x.dev0.tar.gz
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 1,
+\"key_size\": 1,
+\"value_size\":1,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+" \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+"""
+
+from __future__ import absolute_import
+
+import json
+import logging
+import os
+import unittest
+
+import apache_beam as beam
+from apache_beam.pvalue import AsIter
+from apache_beam.testing import synthetic_pipeline
+from apache_beam.testing.load_tests.load_test_metrics_utils import MeasureTime
+from apache_beam.testing.load_tests.load_test_metrics_utils import
MetricsMonitor
+from apache_beam.testing.test_pipeline import TestPipeline
+
+load_test_enabled = False
+if os.environ.get('LOAD_TEST_ENABLED') == 'true':
+ load_test_enabled = True
+
+RUNTIME_LABEL = 'runtime'
+
+
+@unittest.skipIf(not load_test_enabled, 'Enabled only for phase triggering.')
+class SideInputTest(unittest.TestCase):
+ def _parseTestPipelineOptions(self):
+return {
+'numRecords': self.inputOptions.get('num_records'),
+'keySizeBytes': self.inputOptions.get('key_size'),
+'valueSizeBytes': self.inputOptions.get('value_size'),
+'bundleSizeDistribution': {
+'type': self.inputOptions.get(
+'bundle_size_distribution_type', 'const'
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=177131=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-177131 ] ASF GitHub Bot logged work on BEAM-5993: Author: ASF GitHub Bot Created on: 19/Dec/18 18:50 Start Date: 19/Dec/18 18:50 Worklog Time Spent: 10m Work Description: pabloem commented on issue #7020: BEAM-5993 Create SideInput Load test URL: https://github.com/apache/beam/pull/7020#issuecomment-448704153 Thanks Kasia. Sorry that it was just the flakiness. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 177131) Time Spent: 1.5h (was: 1h 20m) > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 1.5h > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=176881=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-176881 ] ASF GitHub Bot logged work on BEAM-5993: Author: ASF GitHub Bot Created on: 19/Dec/18 07:57 Start Date: 19/Dec/18 07:57 Worklog Time Spent: 10m Work Description: kkucharc commented on issue #7020: BEAM-5993 Create SideInput Load test URL: https://github.com/apache/beam/pull/7020#issuecomment-448502713 Run Python PreCommit This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 176881) Time Spent: 1h 20m (was: 1h 10m) > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 1h 20m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=173833=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-173833 ] ASF GitHub Bot logged work on BEAM-5993: Author: ASF GitHub Bot Created on: 11/Dec/18 00:34 Start Date: 11/Dec/18 00:34 Worklog Time Spent: 10m Work Description: pabloem commented on issue #7020: BEAM-5993 Create SideInput Load test URL: https://github.com/apache/beam/pull/7020#issuecomment-446028005 This looks fine to me. I'm glad to merge it. I believe you've run the pipeline right? This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 173833) Time Spent: 1h (was: 50m) > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 1h > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[
https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=173612=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-173612
]
ASF GitHub Bot logged work on BEAM-5993:
Author: ASF GitHub Bot
Created on: 10/Dec/18 14:30
Start Date: 10/Dec/18 14:30
Worklog Time Spent: 10m
Work Description: kkucharc commented on a change in pull request #7020:
BEAM-5993 Create SideInput Load test
URL: https://github.com/apache/beam/pull/7020#discussion_r240230861
##
File path: sdks/python/apache_beam/testing/load_tests/sideinput_test.py
##
@@ -0,0 +1,158 @@
+#
+# 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.
+#
+"""
+To run test on DirectRunner
+
+python setup.py nosetests \
+--test-pipeline-options="
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 300,
+\"key_size\": 5,
+\"value_size\":15,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+ " \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+To run test on other runner (ex. Dataflow):
+
+python setup.py nosetests \
+--test-pipeline-options="
+--runner=TestDataflowRunner
+--project=...
+--staging_location=gs://...
+--temp_location=gs://...
+--sdk_location=./dist/apache-beam-x.x.x.dev0.tar.gz
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 1,
+\"key_size\": 1,
+\"value_size\":1,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+" \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+"""
+
+from __future__ import absolute_import
+
+import json
+import logging
+import unittest
+
+import apache_beam as beam
+from apache_beam.pvalue import AsIter
+from apache_beam.testing import synthetic_pipeline
+from apache_beam.testing.load_tests.load_test_metrics_utils import MeasureTime
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class SideInputTest(unittest.TestCase):
+ def _parseTestPipelineOptions(self):
+return {
+'numRecords': self.inputOptions.get('num_records'),
+'keySizeBytes': self.inputOptions.get('key_size'),
+'valueSizeBytes': self.inputOptions.get('value_size'),
+'bundleSizeDistribution': {
+'type': self.inputOptions.get(
+'bundle_size_distribution_type', 'const'
+),
+'param': self.inputOptions.get('bundle_size_distribution_param', 0)
+},
+'forceNumInitialBundles': self.inputOptions.get(
+'force_initial_num_bundles', 0
+)
+}
+
+ def _getSideInput(self):
+side_input = self._parseTestPipelineOptions()
+side_input['numRecords'] = side_input['numRecords']
+side_input['keySizeBytes'] = side_input['keySizeBytes']
+side_input['valueSizeBytes'] = side_input['valueSizeBytes']
+return side_input
+
+ def _getPerElementDelaySec(self):
+return self.syntheticStepOptions.get('per_element_delay_sec', 0)
+
+ def _getPerBundleDelaySec(self):
+return self.syntheticStepOptions.get('per_bundle_delay_sec', 0)
+
+ def _getOutputRecordsPerInputRecords(self):
+return self.syntheticStepOptions.get('output_records_per_input_records', 0)
+
+ def setUp(self):
+self.pipeline = TestPipeline()
+self.inputOptions = json.loads(self.pipeline.get_option('input_options'))
+self.iterations = self.pipeline.get_option('number_of_counter_operations')
+if self.iterations is None:
+ self.iterations = 1
+
+ class Consume(beam.DoFn):
+def process(self, element, *args, **kwargs):
+ values_list = []
+ for i in element:
+for _, value in i.iteritems():
+ values_list.append(value)
+ yield values_list
+
+ def testSideInput(self):
+def join_fn(element, side_input, iterations):
+ list = []
+ for i in range(iterations):
+for key, value in side_input:
+ if i == iterations - 1:
+
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[
https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=172475=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-172475
]
ASF GitHub Bot logged work on BEAM-5993:
Author: ASF GitHub Bot
Created on: 05/Dec/18 20:59
Start Date: 05/Dec/18 20:59
Worklog Time Spent: 10m
Work Description: pabloem commented on a change in pull request #7020:
BEAM-5993 Create SideInput Load test
URL: https://github.com/apache/beam/pull/7020#discussion_r239234878
##
File path: sdks/python/apache_beam/testing/load_tests/sideinput_test.py
##
@@ -0,0 +1,158 @@
+#
+# 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.
+#
+"""
+To run test on DirectRunner
+
+python setup.py nosetests \
+--test-pipeline-options="
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 300,
+\"key_size\": 5,
+\"value_size\":15,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+ " \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+To run test on other runner (ex. Dataflow):
+
+python setup.py nosetests \
+--test-pipeline-options="
+--runner=TestDataflowRunner
+--project=...
+--staging_location=gs://...
+--temp_location=gs://...
+--sdk_location=./dist/apache-beam-x.x.x.dev0.tar.gz
+--number_of_counter_operations=1000
+--input_options='{
+\"num_records\": 1,
+\"key_size\": 1,
+\"value_size\":1,
+\"bundle_size_distribution_type\": \"const\",
+\"bundle_size_distribution_param\": 1,
+\"force_initial_num_bundles\": 0
+}'
+" \
+--tests apache_beam.testing.load_tests.sideinput_test
+
+"""
+
+from __future__ import absolute_import
+
+import json
+import logging
+import unittest
+
+import apache_beam as beam
+from apache_beam.pvalue import AsIter
+from apache_beam.testing import synthetic_pipeline
+from apache_beam.testing.load_tests.load_test_metrics_utils import MeasureTime
+from apache_beam.testing.test_pipeline import TestPipeline
+
+
+class SideInputTest(unittest.TestCase):
+ def _parseTestPipelineOptions(self):
+return {
+'numRecords': self.inputOptions.get('num_records'),
+'keySizeBytes': self.inputOptions.get('key_size'),
+'valueSizeBytes': self.inputOptions.get('value_size'),
+'bundleSizeDistribution': {
+'type': self.inputOptions.get(
+'bundle_size_distribution_type', 'const'
+),
+'param': self.inputOptions.get('bundle_size_distribution_param', 0)
+},
+'forceNumInitialBundles': self.inputOptions.get(
+'force_initial_num_bundles', 0
+)
+}
+
+ def _getSideInput(self):
+side_input = self._parseTestPipelineOptions()
+side_input['numRecords'] = side_input['numRecords']
+side_input['keySizeBytes'] = side_input['keySizeBytes']
+side_input['valueSizeBytes'] = side_input['valueSizeBytes']
+return side_input
+
+ def _getPerElementDelaySec(self):
+return self.syntheticStepOptions.get('per_element_delay_sec', 0)
+
+ def _getPerBundleDelaySec(self):
+return self.syntheticStepOptions.get('per_bundle_delay_sec', 0)
+
+ def _getOutputRecordsPerInputRecords(self):
+return self.syntheticStepOptions.get('output_records_per_input_records', 0)
+
+ def setUp(self):
+self.pipeline = TestPipeline()
+self.inputOptions = json.loads(self.pipeline.get_option('input_options'))
+self.iterations = self.pipeline.get_option('number_of_counter_operations')
+if self.iterations is None:
+ self.iterations = 1
+
+ class Consume(beam.DoFn):
+def process(self, element, *args, **kwargs):
+ values_list = []
+ for i in element:
+for _, value in i.iteritems():
+ values_list.append(value)
+ yield values_list
+
+ def testSideInput(self):
+def join_fn(element, side_input, iterations):
+ list = []
+ for i in range(iterations):
+for key, value in side_input:
+ if i == iterations - 1:
+
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=171023=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-171023 ] ASF GitHub Bot logged work on BEAM-5993: Author: ASF GitHub Bot Created on: 30/Nov/18 09:02 Start Date: 30/Nov/18 09:02 Worklog Time Spent: 10m Work Description: kkucharc commented on issue #7020: BEAM-5993 Create SideInput Load test URL: https://github.com/apache/beam/pull/7020#issuecomment-443137216 @pabloem I would really appreciate it! I think there is a lot to change, but I'm wondering what's your opinion. This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 171023) Time Spent: 0.5h (was: 20m) > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 0.5h > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Work logged] (BEAM-5993) Create SideInput Load test
[ https://issues.apache.org/jira/browse/BEAM-5993?focusedWorklogId=170873=com.atlassian.jira.plugin.system.issuetabpanels:worklog-tabpanel#worklog-170873 ] ASF GitHub Bot logged work on BEAM-5993: Author: ASF GitHub Bot Created on: 29/Nov/18 23:41 Start Date: 29/Nov/18 23:41 Worklog Time Spent: 10m Work Description: pabloem commented on issue #7020: BEAM-5993 Create SideInput Load test URL: https://github.com/apache/beam/pull/7020#issuecomment-443035531 Do you need a review for this? : ) This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org Issue Time Tracking --- Worklog Id: (was: 170873) Time Spent: 20m (was: 10m) > Create SideInput Load test > -- > > Key: BEAM-5993 > URL: https://issues.apache.org/jira/browse/BEAM-5993 > Project: Beam > Issue Type: Sub-task > Components: testing >Reporter: Kasia Kucharczyk >Assignee: Kasia Kucharczyk >Priority: Major > Time Spent: 20m > Remaining Estimate: 0h > -- This message was sent by Atlassian JIRA (v7.6.3#76005)
