You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by cc...@apache.org on 2018/10/13 00:28:03 UTC
[beam] branch revert-6675-revert-options created (now b152b52)
This is an automated email from the ASF dual-hosted git repository.
ccy pushed a change to branch revert-6675-revert-options
in repository https://gitbox.apache.org/repos/asf/beam.git.
at b152b52 Revert "Revert PRs #6557 #6589 #6600"
This branch includes the following new commits:
new b152b52 Revert "Revert PRs #6557 #6589 #6600"
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
[beam] 01/01: Revert "Revert PRs #6557 #6589 #6600"
Posted by cc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
ccy pushed a commit to branch revert-6675-revert-options
in repository https://gitbox.apache.org/repos/asf/beam.git
commit b152b52a8356969648f493afdb6d8c4db7c6a396
Author: Charles Chen <ch...@users.noreply.github.com>
AuthorDate: Fri Oct 12 17:27:52 2018 -0700
Revert "Revert PRs #6557 #6589 #6600"
---
.../python/apache_beam/options/pipeline_options.py | 32 +++++++++++++++++++++-
.../apache_beam/options/pipeline_options_test.py | 23 ++++++++++++++--
.../runners/worker/sdk_worker_main_test.py | 4 +--
3 files changed, 53 insertions(+), 6 deletions(-)
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index a0059db..a172535 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -20,8 +20,10 @@
from __future__ import absolute_import
import argparse
+import logging
from builtins import list
from builtins import object
+from collections import OrderedDict
from apache_beam.options.value_provider import RuntimeValueProvider
from apache_beam.options.value_provider import StaticValueProvider
@@ -213,7 +215,35 @@ class PipelineOptions(HasDisplayData):
subset[str(cls)] = cls
for cls in subset.values():
cls._add_argparse_args(parser) # pylint: disable=protected-access
- known_args, _ = parser.parse_known_args(self._flags)
+ known_args, unknown_args = parser.parse_known_args(self._flags)
+ # Parse args which are not known at this point but might be recognized
+ # at a later point in time, i.e. by the actual Runner.
+ if unknown_args and unknown_args[0] != '':
+ logging.info("Parsing unknown args: %s", unknown_args)
+
+ def enumerate_args(args):
+ cleaned_args = OrderedDict()
+ for arg in args:
+ if arg.startswith('--'):
+ # split argument name if it's in arg_name=value syntax
+ arg_name = arg.split('=', 1)[0]
+ # count identical arg names
+ if arg_name not in cleaned_args:
+ cleaned_args[arg_name] = 1
+ else:
+ cleaned_args[arg_name] += 1
+ return cleaned_args
+
+ for arg_name, num_times in enumerate_args(unknown_args).items():
+ parser.add_argument(arg_name,
+ nargs='?',
+ action='append' if num_times > 1 else 'store')
+
+ # repeat parsing with unknown options added
+ known_args, unknown_args = parser.parse_known_args(self._flags)
+ if unknown_args:
+ logging.warning("Discarding unparseable args: %s", unknown_args)
+
result = vars(known_args)
# Apply the overrides if any
diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py b/sdks/python/apache_beam/options/pipeline_options_test.py
index 9c14c25..6599753 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -45,16 +45,24 @@ class PipelineOptionsTest(unittest.TestCase):
'display_data': [DisplayDataItemMatcher('num_workers', 5)]},
{
'flags': [
- '--profile_cpu', '--profile_location', 'gs://bucket/', 'ignored'],
+ '--profile_cpu', '--profile_location', 'gs://bucket/',
+ 'ignored', '-invalid=arg', '--unknown_arg', 'unknown_value',
+ '--unknown_flag'
+ ],
'expected': {
'profile_cpu': True, 'profile_location': 'gs://bucket/',
'mock_flag': False, 'mock_option': None,
- 'mock_multi_option': None},
+ 'mock_multi_option': None,
+ 'unknown_arg': 'unknown_value',
+ 'unknown_flag': None},
'display_data': [
DisplayDataItemMatcher('profile_cpu',
True),
DisplayDataItemMatcher('profile_location',
- 'gs://bucket/')]
+ 'gs://bucket/'),
+ DisplayDataItemMatcher('unknown_arg',
+ 'unknown_value')
+ ]
},
{'flags': ['--num_workers', '5', '--mock_flag'],
'expected': {'num_workers': 5,
@@ -281,6 +289,15 @@ class PipelineOptionsTest(unittest.TestCase):
with self.assertRaises(RuntimeError):
options.pot_non_vp_arg1.get()
+ # Converts duplicate unknown argument values to a single argument
+ # with a list value.
+ def test_unknown_duplicate_args_converted_to_list(self):
+ options = PipelineOptions(['--dup_arg', 'val1',
+ '--dup_arg', 'val2',
+ '--dup_arg=val3'])
+ self.assertEqual(options.get_all_options()['dup_arg'],
+ ['val1', 'val2', 'val3'])
+
if __name__ == '__main__':
logging.getLogger().setLevel(logging.INFO)
diff --git a/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py b/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
index 6b5972e..e9b584a 100644
--- a/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
+++ b/sdks/python/apache_beam/runners/worker/sdk_worker_main_test.py
@@ -56,8 +56,8 @@ class SdkWorkerMainTest(unittest.TestCase):
def test_work_count_default_value(self):
self._check_worker_count('{}', 12)
- def test_parse_pipeine_options(self):
- expected_options = PipelineOptions()
+ def test_parse_pipeline_options(self):
+ expected_options = PipelineOptions([])
expected_options.view_as(
SdkWorkerMainTest.MockOptions).m_m_option = [
'worker_threads=1', 'beam_fn_api'