Add experimental ServiceAccount option
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/60907b21 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/60907b21 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/60907b21 Branch: refs/heads/apex-runner Commit: 60907b2128acdae28ecd5fe80d3b00a406b58892 Parents: 0ec956d Author: sammcveety <sam.mcve...@gmail.com> Authored: Sat Oct 22 10:27:53 2016 -0400 Committer: Davor Bonaci <da...@google.com> Committed: Tue Oct 25 13:05:16 2016 -0700 ---------------------------------------------------------------------- pom.xml | 2 +- .../runners/dataflow/DataflowPipelineTranslator.java | 4 ++++ .../dataflow/options/DataflowPipelineOptions.java | 12 ++++++++++++ 3 files changed, 17 insertions(+), 1 deletion(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/60907b21/pom.xml ---------------------------------------------------------------------- diff --git a/pom.xml b/pom.xml index df388d4..307e43e 100644 --- a/pom.xml +++ b/pom.xml @@ -106,7 +106,7 @@ <cloudresourcemanager.version>v1-rev6-1.22.0</cloudresourcemanager.version> <pubsubgrpc.version>0.1.0</pubsubgrpc.version> <clouddebugger.version>v2-rev8-1.22.0</clouddebugger.version> - <dataflow.version>v1b3-rev36-1.22.0</dataflow.version> + <dataflow.version>v1b3-rev42-1.22.0</dataflow.version> <dataflow.proto.version>0.5.160222</dataflow.proto.version> <datastore.client.version>1.2.0</datastore.client.version> <datastore.proto.version>1.2.0</datastore.proto.version> http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/60907b21/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java ---------------------------------------------------------------------- diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java index ad49af7..12aa696 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowPipelineTranslator.java @@ -479,6 +479,10 @@ public class DataflowPipelineTranslator { workerPools.add(workerPool); environment.setWorkerPools(workerPools); + if (options.getServiceAccount() != null) { + environment.setServiceAccountEmail(options.getServiceAccount()); + } + pipeline.traverseTopologically(this); return job; } http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/60907b21/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java ---------------------------------------------------------------------- diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java index ec4fec6..e853f22 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/options/DataflowPipelineOptions.java @@ -22,6 +22,7 @@ import static com.google.common.base.Strings.isNullOrEmpty; import java.io.IOException; import org.apache.beam.runners.dataflow.DataflowRunner; +import org.apache.beam.sdk.annotations.Experimental; import org.apache.beam.sdk.options.ApplicationNameOptions; import org.apache.beam.sdk.options.BigQueryOptions; import org.apache.beam.sdk.options.Default; @@ -29,6 +30,7 @@ import org.apache.beam.sdk.options.DefaultValueFactory; import org.apache.beam.sdk.options.Description; import org.apache.beam.sdk.options.GcpOptions; import org.apache.beam.sdk.options.GcsOptions; +import org.apache.beam.sdk.options.Hidden; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.PubsubOptions; import org.apache.beam.sdk.options.StreamingOptions; @@ -80,6 +82,16 @@ public interface DataflowPipelineOptions void setUpdate(boolean value); /** + * Run the job as a specific service account, instead of the default GCE robot. + */ + @Hidden + @Experimental + @Description( + "Run the job as a specific service account, instead of the default GCE robot.") + String getServiceAccount(); + void setServiceAccount(String value); + + /** * Returns a default staging location under {@link GcpOptions#getGcpTempLocation}. */ class StagingLocationFactory implements DefaultValueFactory<String> {