Repository: incubator-beam Updated Branches: refs/heads/python-sdk afa5ebc70 -> fe1f39609
Add support for experiments Adds a new --experiment flag, to allow runners to optionally enable experimental features. This is similar to the same flag in the java sdk. Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/b2c5f3a8 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/b2c5f3a8 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/b2c5f3a8 Branch: refs/heads/python-sdk Commit: b2c5f3a8f227bdb2cf0092c489701f4e1d762b7b Parents: afa5ebc Author: Ahmet Altay <[email protected]> Authored: Mon Oct 17 18:17:24 2016 -0700 Committer: Robert Bradshaw <[email protected]> Committed: Tue Oct 18 12:04:58 2016 -0700 ---------------------------------------------------------------------- sdks/python/apache_beam/internal/apiclient.py | 6 ++++++ sdks/python/apache_beam/utils/options.py | 9 +++++++++ sdks/python/apache_beam/utils/pipeline_options_test.py | 10 +++++++++- 3 files changed, 24 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/b2c5f3a8/sdks/python/apache_beam/internal/apiclient.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/internal/apiclient.py b/sdks/python/apache_beam/internal/apiclient.py index 3f82f29..8c7cc29 100644 --- a/sdks/python/apache_beam/internal/apiclient.py +++ b/sdks/python/apache_beam/internal/apiclient.py @@ -37,6 +37,7 @@ from apache_beam.utils import retry from apache_beam.utils.dependency import get_required_container_version from apache_beam.utils.dependency import get_sdk_name_and_version from apache_beam.utils.names import PropertyNames +from apache_beam.utils.options import DebugOptions from apache_beam.utils.options import GoogleCloudOptions from apache_beam.utils.options import StandardOptions from apache_beam.utils.options import WorkerOptions @@ -107,6 +108,7 @@ class Environment(object): self.standard_options = options.view_as(StandardOptions) self.google_cloud_options = options.view_as(GoogleCloudOptions) self.worker_options = options.view_as(WorkerOptions) + self.debug_options = options.view_as(DebugOptions) self.proto = dataflow.Environment() self.proto.clusterManagerApiService = COMPUTE_API_SERVICE self.proto.dataset = '%s/cloud_dataflow' % BIGQUERY_API_SERVICE @@ -141,6 +143,10 @@ class Environment(object): value=to_json_value(job_type)), dataflow.Environment.VersionValue.AdditionalProperty( key='major', value=to_json_value(environment_version))]) + # Experiments + if self.debug_options.experiments: + for experiment in self.debug_options.experiments: + self.proto.experiments.append(experiment) # Worker pool(s) information. package_descriptors = [] for package in packages: http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/b2c5f3a8/sdks/python/apache_beam/utils/options.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/utils/options.py b/sdks/python/apache_beam/utils/options.py index a3cea1e..4a56ee2 100644 --- a/sdks/python/apache_beam/utils/options.py +++ b/sdks/python/apache_beam/utils/options.py @@ -364,6 +364,15 @@ class DebugOptions(PipelineOptions): parser.add_argument('--dataflow_job_file', default=None, help='Debug file to write the workflow specification.') + parser.add_argument( + '--experiment', + dest='experiments', + action='append', + default=None, + help= + ('Runners may provide a number of experimental features that can be ' + 'enabled with this flag. Please sync with the owners of the runner ' + 'before enabling any experiments.')) class ProfilingOptions(PipelineOptions): http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/b2c5f3a8/sdks/python/apache_beam/utils/pipeline_options_test.py ---------------------------------------------------------------------- diff --git a/sdks/python/apache_beam/utils/pipeline_options_test.py b/sdks/python/apache_beam/utils/pipeline_options_test.py index 249aa04..ddeecaf 100644 --- a/sdks/python/apache_beam/utils/pipeline_options_test.py +++ b/sdks/python/apache_beam/utils/pipeline_options_test.py @@ -80,7 +80,7 @@ class PipelineOptionsTest(unittest.TestCase): PipelineOptionsTest.MockOptions).mock_option, case['expected']['mock_option']) - def test_option_with_spcae(self): + def test_option_with_space(self): options = PipelineOptions(flags=['--option with space= value with space']) self.assertEqual( getattr(options.view_as(PipelineOptionsTest.MockOptions), @@ -101,6 +101,14 @@ class PipelineOptionsTest(unittest.TestCase): self.assertEqual(options.get_all_options()['num_workers'], 5) self.assertEqual(options.get_all_options()['mock_flag'], True) + def test_experiments(self): + options = PipelineOptions(['--experiment', 'abc', '--experiment', 'def']) + self.assertEqual( + sorted(options.get_all_options()['experiments']), ['abc', 'def']) + + options = PipelineOptions(flags=['']) + self.assertEqual(options.get_all_options()['experiments'], None) + if __name__ == '__main__': logging.getLogger().setLevel(logging.INFO)
