You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by al...@apache.org on 2019/01/29 19:07:49 UTC
[beam] branch master updated: [BEAM-5442] Retrieve valid runner
options from JobService
This is an automated email from the ASF dual-hosted git repository.
altay 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 a991c1c [BEAM-5442] Retrieve valid runner options from JobService
new 1a6490d Merge pull request #7597 from tweise/BEAM-5442.runnerOptions
a991c1c is described below
commit a991c1c93d623473b8e2dd5c756bf495c9953d26
Author: Thomas Weise <th...@apache.org>
AuthorDate: Tue Jan 22 21:28:59 2019 -0800
[BEAM-5442] Retrieve valid runner options from JobService
---
.../src/main/proto/beam_job_api.proto | 48 +++++++++++
.../jobsubmission/InMemoryJobService.java | 22 ++++++
sdks/java/core/build.gradle | 1 +
.../beam/sdk/options/PipelineOptionsFactory.java | 84 ++++++++++++++++++++
.../sdk/options/PipelineOptionsFactoryTest.java | 92 ++++++++++++++++++++++
.../python/apache_beam/options/pipeline_options.py | 67 ++++------------
.../apache_beam/options/pipeline_options_test.py | 46 ++++-------
.../runners/portability/flink_runner_test.py | 5 +-
.../runners/portability/local_job_service.py | 3 +
.../runners/portability/portable_runner.py | 52 +++++++++---
10 files changed, 323 insertions(+), 97 deletions(-)
diff --git a/model/job-management/src/main/proto/beam_job_api.proto b/model/job-management/src/main/proto/beam_job_api.proto
index 80d6f40..a56b322 100644
--- a/model/job-management/src/main/proto/beam_job_api.proto
+++ b/model/job-management/src/main/proto/beam_job_api.proto
@@ -54,6 +54,9 @@ service JobService {
// Subscribe to a stream of state changes and messages from the job
rpc GetMessageStream (JobMessagesRequest) returns (stream JobMessagesResponse);
+
+ // Get the supported pipeline options of the runner
+ rpc DescribePipelineOptions (DescribePipelineOptionsRequest) returns (DescribePipelineOptionsResponse);
}
@@ -176,3 +179,48 @@ message JobState {
CANCELLING = 10;
}
}
+
+
+// DescribePipelineOptions provides metadata about the options supported by a runner.
+// It will be used by the SDK client to validate the options specified by or
+// list available options to the user.
+// Throws error GRPC_STATUS_UNAVAILABLE if server is down
+message DescribePipelineOptionsRequest {
+}
+
+// Type for pipeline options.
+// Types mirror those of JSON, since that's how pipeline options are serialized.
+message PipelineOptionType {
+ enum Enum {
+ STRING = 0;
+ BOOLEAN = 1;
+ // whole numbers, see https://json-schema.org/understanding-json-schema/reference/numeric.html
+ INTEGER = 2;
+ NUMBER = 3;
+ ARRAY = 4;
+ OBJECT = 5;
+ };
+}
+
+// Metadata for a pipeline option.
+message PipelineOptionDescriptor {
+ // (Required) The option name.
+ string name = 1;
+
+ // (Required) Type of option.
+ PipelineOptionType.Enum type = 2;
+
+ // (Optional) Description suitable for display / help text.
+ string description = 3;
+
+ // (Optional) Default value.
+ string default_value = 4;
+
+ // (Required) The group this option belongs to.
+ string group = 5;
+}
+
+message DescribePipelineOptionsResponse {
+ // List of pipeline option descriptors.
+ repeated PipelineOptionDescriptor options = 1;
+}
diff --git a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java
index 3ea4939..906936e 100644
--- a/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java
+++ b/runners/java-fn-execution/src/main/java/org/apache/beam/runners/fnexecution/jobsubmission/InMemoryJobService.java
@@ -24,6 +24,8 @@ import java.util.function.Consumer;
import java.util.function.Function;
import org.apache.beam.model.jobmanagement.v1.JobApi.CancelJobRequest;
import org.apache.beam.model.jobmanagement.v1.JobApi.CancelJobResponse;
+import org.apache.beam.model.jobmanagement.v1.JobApi.DescribePipelineOptionsRequest;
+import org.apache.beam.model.jobmanagement.v1.JobApi.DescribePipelineOptionsResponse;
import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateRequest;
import org.apache.beam.model.jobmanagement.v1.JobApi.GetJobStateResponse;
import org.apache.beam.model.jobmanagement.v1.JobApi.JobMessage;
@@ -40,6 +42,7 @@ import org.apache.beam.runners.core.construction.graph.PipelineValidator;
import org.apache.beam.runners.fnexecution.FnService;
import org.apache.beam.sdk.fn.function.ThrowingConsumer;
import org.apache.beam.sdk.fn.stream.SynchronizedStreamObserver;
+import org.apache.beam.sdk.options.PipelineOptionsFactory;
import org.apache.beam.vendor.grpc.v1p13p1.com.google.protobuf.Struct;
import org.apache.beam.vendor.grpc.v1p13p1.io.grpc.Status;
import org.apache.beam.vendor.grpc.v1p13p1.io.grpc.StatusException;
@@ -309,6 +312,25 @@ public class InMemoryJobService extends JobServiceGrpc.JobServiceImplBase implem
}
@Override
+ public void describePipelineOptions(
+ DescribePipelineOptionsRequest request,
+ StreamObserver<DescribePipelineOptionsResponse> responseObserver) {
+ LOG.trace("{} {}", DescribePipelineOptionsRequest.class.getSimpleName(), request);
+ try {
+ DescribePipelineOptionsResponse response =
+ DescribePipelineOptionsResponse.newBuilder()
+ .addAllOptions(
+ PipelineOptionsFactory.describe(PipelineOptionsFactory.getRegisteredOptions()))
+ .build();
+ responseObserver.onNext(response);
+ responseObserver.onCompleted();
+ } catch (Exception e) {
+ LOG.error("Error describing pipeline options", e);
+ responseObserver.onError(Status.INTERNAL.withCause(e).asException());
+ }
+ }
+
+ @Override
public void close() throws Exception {
// TODO: throw error if jobs are running
}
diff --git a/sdks/java/core/build.gradle b/sdks/java/core/build.gradle
index c6d29fc..86fcc95 100644
--- a/sdks/java/core/build.gradle
+++ b/sdks/java/core/build.gradle
@@ -61,6 +61,7 @@ dependencies {
antlr library.java.antlr
// Required to load constants from the model, e.g. max timestamp for global window
shadow project(path: ":beam-model-pipeline", configuration: "shadow")
+ shadow project(path: ":beam-model-job-management", configuration: "shadow")
shadow library.java.vendored_guava_20_0
compile library.java.antlr_runtime
compile library.java.protobuf_java
diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
index 1629b7a..e77b92d 100644
--- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
+++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptionsFactory.java
@@ -35,6 +35,7 @@ import java.lang.reflect.Method;
import java.lang.reflect.Modifier;
import java.lang.reflect.Proxy;
import java.lang.reflect.Type;
+import java.math.BigInteger;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collection;
@@ -55,6 +56,8 @@ import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;
import java.util.stream.StreamSupport;
import javax.annotation.Nonnull;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionDescriptor;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionType;
import org.apache.beam.sdk.PipelineRunner;
import org.apache.beam.sdk.annotations.Experimental;
import org.apache.beam.sdk.options.Validation.Required;
@@ -63,6 +66,7 @@ import org.apache.beam.sdk.transforms.display.DisplayData;
import org.apache.beam.sdk.util.StringUtils;
import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.apache.beam.vendor.guava.v20_0.com.google.common.annotations.VisibleForTesting;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.base.CaseFormat;
import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Function;
import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Joiner;
import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Optional;
@@ -630,6 +634,86 @@ public class PipelineOptionsFactory {
}
}
+ private static final Set<Class<?>> JSON_INTEGER_TYPES =
+ Sets.newHashSet(
+ short.class,
+ Short.class,
+ int.class,
+ Integer.class,
+ long.class,
+ Long.class,
+ BigInteger.class);
+
+ private static final Set<Class<?>> JSON_NUMBER_TYPES =
+ Sets.newHashSet(
+ float.class, Float.class, double.class, Double.class, java.math.BigDecimal.class);
+
+ /**
+ * Outputs the set of options available to be set for the passed in {@link PipelineOptions}
+ * interfaces. The output for consumption of the job service client.
+ */
+ public static List<PipelineOptionDescriptor> describe(
+ Set<Class<? extends PipelineOptions>> ifaces) {
+ checkNotNull(ifaces);
+ List<PipelineOptionDescriptor> result = new ArrayList<>();
+ Set<Method> seenMethods = Sets.newHashSet();
+
+ for (Class<? extends PipelineOptions> iface : ifaces) {
+ CACHE.get().validateWellFormed(iface);
+
+ Set<PipelineOptionSpec> properties = PipelineOptionsReflector.getOptionSpecs(iface);
+
+ RowSortedTable<Class<?>, String, Method> ifacePropGetterTable =
+ TreeBasedTable.create(ClassNameComparator.INSTANCE, Ordering.natural());
+ for (PipelineOptionSpec prop : properties) {
+ ifacePropGetterTable.put(
+ prop.getDefiningInterface(), prop.getName(), prop.getGetterMethod());
+ }
+
+ for (Map.Entry<Class<?>, Map<String, Method>> ifaceToPropertyMap :
+ ifacePropGetterTable.rowMap().entrySet()) {
+ Class<?> currentIface = ifaceToPropertyMap.getKey();
+ Map<String, Method> propertyNamesToGetters = ifaceToPropertyMap.getValue();
+
+ List<String> lists = Lists.newArrayList(propertyNamesToGetters.keySet());
+ lists.sort(String.CASE_INSENSITIVE_ORDER);
+ for (String propertyName : lists) {
+ Method method = propertyNamesToGetters.get(propertyName);
+ if (!seenMethods.add(method)) {
+ continue;
+ }
+ Class<?> returnType = method.getReturnType();
+ PipelineOptionType.Enum optionType = PipelineOptionType.Enum.STRING;
+ if (JSON_INTEGER_TYPES.contains(returnType)) {
+ optionType = PipelineOptionType.Enum.INTEGER;
+ } else if (JSON_NUMBER_TYPES.contains(returnType)) {
+ optionType = PipelineOptionType.Enum.NUMBER;
+ } else if (returnType == boolean.class || returnType == Boolean.class) {
+ optionType = PipelineOptionType.Enum.BOOLEAN;
+ } else if (List.class.isAssignableFrom(returnType)) {
+ optionType = PipelineOptionType.Enum.ARRAY;
+ }
+ String optionName = CaseFormat.LOWER_CAMEL.to(CaseFormat.LOWER_UNDERSCORE, propertyName);
+ Description description = method.getAnnotation(Description.class);
+ PipelineOptionDescriptor.Builder builder =
+ PipelineOptionDescriptor.newBuilder()
+ .setName(optionName)
+ .setType(optionType)
+ .setGroup(currentIface.getName());
+ Optional<String> defaultValue = getDefaultValueFromAnnotation(method);
+ if (defaultValue.isPresent()) {
+ builder.setDefaultValue(defaultValue.get());
+ }
+ if (description != null) {
+ builder.setDescription(description.value());
+ }
+ result.add(builder.build());
+ }
+ }
+ }
+ return result;
+ }
+
/**
* Output the requirement groups that the property is a member of, including all properties that
* satisfy the group requirement, breaking up long lines on white space characters and attempting
diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
index 8e96d25..5168a67 100644
--- a/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
+++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/options/PipelineOptionsFactoryTest.java
@@ -18,10 +18,13 @@
package org.apache.beam.sdk.options;
import static java.util.Locale.ROOT;
+import static org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Maps.uniqueIndex;
import static org.hamcrest.Matchers.containsString;
import static org.hamcrest.Matchers.equalTo;
import static org.hamcrest.Matchers.hasItem;
+import static org.hamcrest.Matchers.isEmptyString;
import static org.hamcrest.Matchers.not;
+import static org.hamcrest.Matchers.notNullValue;
import static org.hamcrest.Matchers.nullValue;
import static org.junit.Assert.assertArrayEquals;
import static org.junit.Assert.assertEquals;
@@ -46,12 +49,16 @@ import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
import com.fasterxml.jackson.databind.annotation.JsonSerialize;
import com.fasterxml.jackson.databind.module.SimpleModule;
import com.google.auto.service.AutoService;
+import com.google.common.collect.Sets;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.PrintStream;
+import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionDescriptor;
+import org.apache.beam.model.jobmanagement.v1.JobApi.PipelineOptionType;
import org.apache.beam.sdk.Pipeline;
import org.apache.beam.sdk.PipelineResult;
import org.apache.beam.sdk.PipelineRunner;
@@ -63,6 +70,7 @@ import org.apache.beam.sdk.testing.RestoreSystemProperties;
import org.apache.beam.sdk.util.common.ReflectHelpers;
import org.apache.beam.vendor.guava.v20_0.com.google.common.base.Charsets;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ArrayListMultimap;
+import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.Collections2;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableList;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ImmutableMap;
import org.apache.beam.vendor.guava.v20_0.com.google.common.collect.ListMultimap;
@@ -2049,4 +2057,88 @@ public class PipelineOptionsFactoryTest {
return o.getMyMethod();
}
}
+
+ /** Test interface. */
+ public interface TestDescribeOptions extends PipelineOptions {
+ String getString();
+
+ void setString(String value);
+
+ @Description("integer property")
+ Integer getInteger();
+
+ void setInteger(Integer value);
+
+ @Description("float number property")
+ Float getFloat();
+
+ void setFloat(Float value);
+
+ @Description("simple boolean property")
+ @Default.Boolean(true)
+ boolean getBooleanSimple();
+
+ void setBooleanSimple(boolean value);
+
+ @Default.Boolean(false)
+ Boolean getBooleanWrapper();
+
+ void setBooleanWrapper(Boolean value);
+
+ List<Integer> getList();
+
+ void setList(List<Integer> value);
+ }
+
+ @Test
+ public void testDescribe() {
+ List<PipelineOptionDescriptor> described =
+ PipelineOptionsFactory.describe(
+ Sets.newHashSet(PipelineOptions.class, TestDescribeOptions.class));
+
+ Map<String, PipelineOptionDescriptor> mapped = uniqueIndex(described, input -> input.getName());
+ assertEquals("no duplicates", described.size(), mapped.size());
+
+ Collection<PipelineOptionDescriptor> filtered =
+ Collections2.filter(
+ described, input -> input.getGroup().equals(TestDescribeOptions.class.getName()));
+ assertEquals(6, filtered.size());
+ mapped = uniqueIndex(filtered, input -> input.getName());
+
+ PipelineOptionDescriptor listDesc = mapped.get("list");
+ assertThat(listDesc, notNullValue());
+ assertThat(listDesc.getDescription(), isEmptyString());
+ assertEquals(PipelineOptionType.Enum.ARRAY, listDesc.getType());
+ assertThat(listDesc.getDefaultValue(), isEmptyString());
+
+ PipelineOptionDescriptor stringDesc = mapped.get("string");
+ assertThat(stringDesc, notNullValue());
+ assertThat(stringDesc.getDescription(), isEmptyString());
+ assertEquals(PipelineOptionType.Enum.STRING, stringDesc.getType());
+ assertThat(stringDesc.getDefaultValue(), isEmptyString());
+
+ PipelineOptionDescriptor integerDesc = mapped.get("integer");
+ assertThat(integerDesc, notNullValue());
+ assertEquals("integer property", integerDesc.getDescription());
+ assertEquals(PipelineOptionType.Enum.INTEGER, integerDesc.getType());
+ assertThat(integerDesc.getDefaultValue(), isEmptyString());
+
+ PipelineOptionDescriptor floatDesc = mapped.get("float");
+ assertThat(integerDesc, notNullValue());
+ assertEquals("float number property", floatDesc.getDescription());
+ assertEquals(PipelineOptionType.Enum.NUMBER, floatDesc.getType());
+ assertThat(floatDesc.getDefaultValue(), isEmptyString());
+
+ PipelineOptionDescriptor booleanSimpleDesc = mapped.get("boolean_simple");
+ assertThat(booleanSimpleDesc, notNullValue());
+ assertEquals("simple boolean property", booleanSimpleDesc.getDescription());
+ assertEquals(PipelineOptionType.Enum.BOOLEAN, booleanSimpleDesc.getType());
+ assertThat(booleanSimpleDesc.getDefaultValue(), equalTo("true"));
+
+ PipelineOptionDescriptor booleanWrapperDesc = mapped.get("boolean_wrapper");
+ assertThat(booleanWrapperDesc, notNullValue());
+ assertThat(booleanWrapperDesc.getDescription(), isEmptyString());
+ assertEquals(PipelineOptionType.Enum.BOOLEAN, booleanWrapperDesc.getType());
+ assertThat(booleanWrapperDesc.getDefaultValue(), equalTo("false"));
+ }
}
diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py
index e051f78..ce55d9b 100644
--- a/sdks/python/apache_beam/options/pipeline_options.py
+++ b/sdks/python/apache_beam/options/pipeline_options.py
@@ -23,7 +23,6 @@ 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
@@ -201,7 +200,7 @@ class PipelineOptions(HasDisplayData):
return cls(flags)
- def get_all_options(self, drop_default=False):
+ def get_all_options(self, drop_default=False, add_extra_args_fn=None):
"""Returns a dictionary of all defined arguments.
Returns a dictionary of all defined arguments (arguments that are defined in
@@ -210,6 +209,8 @@ class PipelineOptions(HasDisplayData):
Args:
drop_default: If set to true, options that are equal to their default
values, are not returned as part of the result dictionary.
+ add_extra_args_fn: Callback to populate additional arguments, can be used
+ by runner to supply otherwise unknown args.
Returns:
Dictionary of all args and values.
@@ -223,35 +224,11 @@ class PipelineOptions(HasDisplayData):
subset[str(cls)] = cls
for cls in subset.values():
cls._add_argparse_args(parser) # pylint: disable=protected-access
+ if add_extra_args_fn:
+ add_extra_args_fn(parser)
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)
-
+ if unknown_args:
+ logging.warning("Discarding unparseable args: %s", unknown_args)
result = vars(known_args)
# Apply the overrides if any
@@ -733,31 +710,17 @@ class PortableOptions(PipelineOptions):
'command.'))
-class FlinkOptions(PipelineOptions):
+class RunnerOptions(PipelineOptions):
+ """Runner options are provided by the job service.
+ The SDK has no a priori knowledge of runner options.
+ It should be able to work with any portable runner.
+ Runner specific options are discovered from the job service endpoint.
+ """
@classmethod
def _add_argparse_args(cls, parser):
- parser.add_argument('--flink_master',
- type=str,
- help=
- ('Address of the Flink master where the pipeline '
- 'should be executed. Can either be of the form '
- '\'host:port\' or one of the special values '
- '[local], [collection], or [auto].'))
- parser.add_argument('--parallelism',
- type=int,
- help=
- ('The degree of parallelism to be used when '
- 'distributing operations onto workers.'))
- parser.add_argument('--shutdown_sources_on_final_watermark',
- default=False,
- action='store_true',
- help=
- ('If set to true, allows sources to shutdown '
- 'after emitting the final Watermark. '
- 'Note: Checkpoints/Savepoints can only be '
- 'taken when all operators, including sources, '
- 'are running.'))
+ # TODO: help option to display discovered options
+ pass
class TestOptions(PipelineOptions):
diff --git a/sdks/python/apache_beam/options/pipeline_options_test.py b/sdks/python/apache_beam/options/pipeline_options_test.py
index dac9b49..021097b 100644
--- a/sdks/python/apache_beam/options/pipeline_options_test.py
+++ b/sdks/python/apache_beam/options/pipeline_options_test.py
@@ -47,24 +47,16 @@ class PipelineOptionsTest(unittest.TestCase):
'display_data': [DisplayDataItemMatcher('num_workers', 5)]},
{
'flags': [
- '--profile_cpu', '--profile_location', 'gs://bucket/',
- 'ignored', '-invalid=arg', '--unknown_arg', 'unknown_value',
- '--unknown_flag'
- ],
+ '--profile_cpu', '--profile_location', 'gs://bucket/', 'ignored'],
'expected': {
'profile_cpu': True, 'profile_location': 'gs://bucket/',
'mock_flag': False, 'mock_option': None,
- 'mock_multi_option': None,
- 'unknown_arg': 'unknown_value',
- 'unknown_flag': None},
+ 'mock_multi_option': None},
'display_data': [
DisplayDataItemMatcher('profile_cpu',
True),
DisplayDataItemMatcher('profile_location',
- 'gs://bucket/'),
- DisplayDataItemMatcher('unknown_arg',
- 'unknown_value')
- ]
+ 'gs://bucket/')]
},
{'flags': ['--num_workers', '5', '--mock_flag'],
'expected': {'num_workers': 5,
@@ -114,16 +106,6 @@ class PipelineOptionsTest(unittest.TestCase):
'display_data': [
DisplayDataItemMatcher('mock_multi_option', ['op1', 'op2'])]
},
- {'flags': ['--flink_master=testmaster:8081', '--parallelism=42'],
- 'expected': {'flink_master': 'testmaster:8081',
- 'parallelism': 42,
- 'mock_flag': False,
- 'mock_option': None,
- 'mock_multi_option': None},
- 'display_data': [
- DisplayDataItemMatcher('flink_master', 'testmaster:8081'),
- DisplayDataItemMatcher('parallelism', 42)]
- },
]
# Used for testing newly added flags.
@@ -291,14 +273,20 @@ 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'])
+ # Converts extra arguments to list value.
+ def test_extra_args(self):
+ options = PipelineOptions([
+ '--extra_arg', 'val1',
+ '--extra_arg', 'val2',
+ '--extra_arg=val3',
+ '--unknown_arg', 'val4'])
+
+ def add_extra_options(parser):
+ parser.add_argument("--extra_arg", action='append')
+
+ self.assertEqual(options.get_all_options(
+ add_extra_args_fn=add_extra_options)
+ ['extra_arg'], ['val1', 'val2', 'val3'])
# The argparse package by default tries to autocomplete option names. This
# results in an "ambiguous option" error from argparse when an unknown option
diff --git a/sdks/python/apache_beam/runners/portability/flink_runner_test.py b/sdks/python/apache_beam/runners/portability/flink_runner_test.py
index 500ef8e..b43721a 100644
--- a/sdks/python/apache_beam/runners/portability/flink_runner_test.py
+++ b/sdks/python/apache_beam/runners/portability/flink_runner_test.py
@@ -30,7 +30,6 @@ from tempfile import mkdtemp
import apache_beam as beam
from apache_beam.metrics import Metrics
from apache_beam.options.pipeline_options import DebugOptions
-from apache_beam.options.pipeline_options import FlinkOptions
from apache_beam.options.pipeline_options import PortableOptions
from apache_beam.options.pipeline_options import StandardOptions
from apache_beam.runners.portability import portable_runner
@@ -129,8 +128,8 @@ if __name__ == '__main__':
def create_options(self):
options = super(FlinkRunnerTest, self).create_options()
options.view_as(DebugOptions).experiments = ['beam_fn_api']
- options.view_as(FlinkOptions).parallelism = 1
- options.view_as(FlinkOptions).shutdown_sources_on_final_watermark = True
+ options._all_options['parallelism'] = 1
+ options._all_options['shutdown_sources_on_final_watermark'] = True
options.view_as(PortableOptions).environment_type = (
environment_type.upper())
if environment_config:
diff --git a/sdks/python/apache_beam/runners/portability/local_job_service.py b/sdks/python/apache_beam/runners/portability/local_job_service.py
index 4610c7a..91ceff7 100644
--- a/sdks/python/apache_beam/runners/portability/local_job_service.py
+++ b/sdks/python/apache_beam/runners/portability/local_job_service.py
@@ -161,6 +161,9 @@ class LocalJobServicer(beam_job_api_pb2_grpc.JobServiceServicer):
resp = beam_job_api_pb2.JobMessagesResponse(message_response=msg)
yield resp
+ def DescribePipelineOptions(self, request, context=None):
+ return beam_job_api_pb2.DescribePipelineOptionsResponse()
+
class SubprocessSdkWorker(object):
"""Manages a SDK worker implemented as a subprocess communicating over grpc.
diff --git a/sdks/python/apache_beam/runners/portability/portable_runner.py b/sdks/python/apache_beam/runners/portability/portable_runner.py
index bb750e5..42bca33 100644
--- a/sdks/python/apache_beam/runners/portability/portable_runner.py
+++ b/sdks/python/apache_beam/runners/portability/portable_runner.py
@@ -189,12 +189,6 @@ class PortableRunner(runner.PipelineRunner):
proto_pipeline = fn_api_runner_transforms.with_stages(
proto_pipeline, stages)
- # TODO: Define URNs for options.
- # convert int values: https://issues.apache.org/jira/browse/BEAM-5509
- p_options = {'beam:option:' + k + ':v1': (str(v) if type(v) == int else v)
- for k, v in options.get_all_options().items()
- if v is not None}
-
if not job_service:
channel = grpc.insecure_channel(job_endpoint)
grpc.channel_ready_future(channel).result()
@@ -202,8 +196,9 @@ class PortableRunner(runner.PipelineRunner):
else:
channel = None
- # Sends the PrepareRequest but retries in case the channel is not ready
- def send_prepare_request(max_retries=5):
+ # fetch runner options from job service
+ # retries in case the channel is not ready
+ def send_options_request(max_retries=5):
num_retries = 0
while True:
try:
@@ -211,16 +206,47 @@ class PortableRunner(runner.PipelineRunner):
# Seems to be only an issue on Mac with port forwardings
if channel:
grpc.channel_ready_future(channel).result()
- return job_service.Prepare(
- beam_job_api_pb2.PrepareJobRequest(
- job_name='job', pipeline=proto_pipeline,
- pipeline_options=job_utils.dict_to_struct(p_options)))
+ return job_service.DescribePipelineOptions(
+ beam_job_api_pb2.DescribePipelineOptionsRequest())
except grpc._channel._Rendezvous as e:
num_retries += 1
if num_retries > max_retries:
raise e
- prepare_response = send_prepare_request()
+ options_response = send_options_request()
+
+ def add_runner_options(parser):
+ for option in options_response.options:
+ try:
+ # no default values - we don't want runner options
+ # added unless they were specified by the user
+ add_arg_args = {'action' : 'store', 'help' : option.description}
+ if option.type == beam_job_api_pb2.PipelineOptionType.BOOLEAN:
+ add_arg_args['action'] = 'store_true'\
+ if option.default_value != 'true' else 'store_false'
+ elif option.type == beam_job_api_pb2.PipelineOptionType.INTEGER:
+ add_arg_args['type'] = int
+ elif option.type == beam_job_api_pb2.PipelineOptionType.ARRAY:
+ add_arg_args['action'] = 'append'
+ parser.add_argument("--%s" % option.name, **add_arg_args)
+ except Exception as e:
+ # ignore runner options that are already present
+ # only in this case is duplicate not treated as error
+ if 'conflicting option string' not in str(e):
+ raise
+ logging.debug("Runner option '%s' was already added" % option.name)
+
+ all_options = options.get_all_options(add_extra_args_fn=add_runner_options)
+ # TODO: Define URNs for options.
+ # convert int values: https://issues.apache.org/jira/browse/BEAM-5509
+ p_options = {'beam:option:' + k + ':v1': (str(v) if type(v) == int else v)
+ for k, v in all_options.items()
+ if v is not None}
+
+ prepare_response = job_service.Prepare(
+ beam_job_api_pb2.PrepareJobRequest(
+ job_name='job', pipeline=proto_pipeline,
+ pipeline_options=job_utils.dict_to_struct(p_options)))
if prepare_response.artifact_staging_endpoint.url:
stager = portable_stager.PortableStager(
grpc.insecure_channel(prepare_response.artifact_staging_endpoint.url),