This is an automated email from the ASF dual-hosted git repository. lcwik pushed a commit to branch master in repository https://gitbox.apache.org/repos/asf/beam.git
commit 57310a695ad49aa7e4c5b9fd65d79b484674f27c Author: Luke Cwik <lc...@google.com> AuthorDate: Wed Dec 20 11:26:09 2017 -0800 Post code review amendments and refactor --- .../org/apache/beam/runners/flink/FlinkRunner.java | 36 +--------- .../beam/runners/dataflow/DataflowRunner.java | 41 ++---------- .../beam/runners/dataflow/DataflowRunnerTest.java | 36 ---------- .../beam/runners/spark/SparkContextOptions.java | 2 +- .../beam/runners/spark/SparkPipelineOptions.java | 21 +++--- .../org/apache/beam/runners/spark/SparkRunner.java | 13 ++++ .../spark/translation/SparkContextFactory.java | 4 +- .../org/apache/beam/sdk/util/PipelineUtils.java | 59 +++++++++++++++++ .../apache/beam/sdk/util/PipelineUtilsTest.java | 76 ++++++++++++++++++++++ 9 files changed, 170 insertions(+), 118 deletions(-) diff --git a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java index ca12615..f2be9a7 100644 --- a/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java +++ b/runners/flink/src/main/java/org/apache/beam/runners/flink/FlinkRunner.java @@ -17,14 +17,11 @@ */ package org.apache.beam.runners.flink; +import static org.apache.beam.sdk.util.PipelineUtils.detectClassPathResourcesToStage; + import com.google.common.base.Joiner; -import java.io.File; -import java.net.URISyntaxException; -import java.net.URL; -import java.net.URLClassLoader; import java.util.ArrayList; import java.util.HashSet; -import java.util.List; import java.util.Map; import java.util.Set; import java.util.SortedSet; @@ -150,36 +147,7 @@ public class FlinkRunner extends PipelineRunner<PipelineResult> { return "FlinkRunner#" + hashCode(); } - /** - * Attempts to detect all the resources the class loader has access to. This does not recurse - * to class loader parents stopping it from pulling in resources from the system class loader. - * - * @param classLoader The URLClassLoader to use to detect resources to stage. - * @return A list of absolute paths to the resources the class loader uses. - * @throws IllegalArgumentException If either the class loader is not a URLClassLoader or one - * of the resources the class loader exposes is not a file resource. - */ - protected static List<String> detectClassPathResourcesToStage( - ClassLoader classLoader) { - if (!(classLoader instanceof URLClassLoader)) { - String message = String.format("Unable to use ClassLoader to detect classpath elements. " - + "Current ClassLoader is %s, only URLClassLoaders are supported.", classLoader); - LOG.error(message); - throw new IllegalArgumentException(message); - } - List<String> files = new ArrayList<>(); - for (URL url : ((URLClassLoader) classLoader).getURLs()) { - try { - files.add(new File(url.toURI()).getAbsolutePath()); - } catch (IllegalArgumentException | URISyntaxException e) { - String message = String.format("Unable to convert url (%s) to file.", url); - LOG.error(message); - throw new IllegalArgumentException(message, e); - } - } - return files; - } /** A set of {@link View}s with non-deterministic key coders. */ Set<PTransform<?, ?>> ptransformViewsWithNonDeterministicKeyCoders; diff --git a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java index 942d36b..729ec9c 100644 --- a/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java +++ b/runners/google-cloud-dataflow-java/src/main/java/org/apache/beam/runners/dataflow/DataflowRunner.java @@ -22,6 +22,7 @@ import static com.google.common.base.Preconditions.checkArgument; import static com.google.common.base.Preconditions.checkState; import static com.google.common.base.Strings.isNullOrEmpty; import static org.apache.beam.sdk.util.CoderUtils.encodeToByteArray; +import static org.apache.beam.sdk.util.PipelineUtils.detectClassPathResourcesToStage; import static org.apache.beam.sdk.util.SerializableUtils.serializeToByteArray; import static org.apache.beam.sdk.util.StringUtils.byteArrayToJsonString; @@ -41,12 +42,8 @@ import com.google.common.base.Utf8; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; import com.google.common.collect.Iterables; -import java.io.File; import java.io.IOException; import java.io.PrintWriter; -import java.net.URISyntaxException; -import java.net.URL; -import java.net.URLClassLoader; import java.nio.channels.Channels; import java.util.ArrayList; import java.util.Arrays; @@ -256,9 +253,9 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> { throw new IllegalArgumentException("No files to stage has been found."); } else { LOG.info("PipelineOptions.filesToStage was not specified. " - + "Defaulting to files from the classpath: will stage {} files. " - + "Enable logging at DEBUG level to see which files will be staged.", - dataflowOptions.getFilesToStage().size()); + + "Defaulting to files from the classpath: will stage {} files. " + + "Enable logging at DEBUG level to see which files will be staged.", + dataflowOptions.getFilesToStage().size()); LOG.debug("Classpath elements: {}", dataflowOptions.getFilesToStage()); } } @@ -1566,36 +1563,6 @@ public class DataflowRunner extends PipelineRunner<DataflowPipelineJob> { } /** - * Attempts to detect all the resources the class loader has access to. This does not recurse - * to class loader parents stopping it from pulling in resources from the system class loader. - * - * @param classLoader The URLClassLoader to use to detect resources to stage. - * @throws IllegalArgumentException If either the class loader is not a URLClassLoader or one - * of the resources the class loader exposes is not a file resource. - * @return A list of absolute paths to the resources the class loader uses. - */ - protected static List<String> detectClassPathResourcesToStage(ClassLoader classLoader) { - if (!(classLoader instanceof URLClassLoader)) { - String message = String.format("Unable to use ClassLoader to detect classpath elements. " - + "Current ClassLoader is %s, only URLClassLoaders are supported.", classLoader); - LOG.error(message); - throw new IllegalArgumentException(message); - } - - List<String> files = new ArrayList<>(); - for (URL url : ((URLClassLoader) classLoader).getURLs()) { - try { - files.add(new File(url.toURI()).getAbsolutePath()); - } catch (IllegalArgumentException | URISyntaxException e) { - String message = String.format("Unable to convert url (%s) to file.", url); - LOG.error(message); - throw new IllegalArgumentException(message, e); - } - } - return files; - } - - /** * Finds the id for the running job of the given name. */ private String getJobIdFromName(String jobName) { diff --git a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java index edf513b..90748af 100644 --- a/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java +++ b/runners/google-cloud-dataflow-java/src/test/java/org/apache/beam/runners/dataflow/DataflowRunnerTest.java @@ -53,8 +53,6 @@ import java.io.File; import java.io.FileNotFoundException; import java.io.IOException; import java.io.Serializable; -import java.net.URL; -import java.net.URLClassLoader; import java.nio.channels.FileChannel; import java.nio.channels.SeekableByteChannel; import java.nio.file.Files; @@ -628,40 +626,6 @@ public class DataflowRunnerTest implements Serializable { } @Test - public void detectClassPathResourceWithFileResources() throws Exception { - File file = tmpFolder.newFile("file"); - File file2 = tmpFolder.newFile("file2"); - URLClassLoader classLoader = new URLClassLoader(new URL[] { - file.toURI().toURL(), - file2.toURI().toURL() - }); - - assertEquals(ImmutableList.of(file.getAbsolutePath(), file2.getAbsolutePath()), - DataflowRunner.detectClassPathResourcesToStage(classLoader)); - } - - @Test - public void detectClassPathResourcesWithUnsupportedClassLoader() { - ClassLoader mockClassLoader = Mockito.mock(ClassLoader.class); - thrown.expect(IllegalArgumentException.class); - thrown.expectMessage("Unable to use ClassLoader to detect classpath elements."); - - DataflowRunner.detectClassPathResourcesToStage(mockClassLoader); - } - - @Test - public void detectClassPathResourceWithNonFileResources() throws Exception { - String url = "http://www.google.com/all-the-secrets.jar"; - URLClassLoader classLoader = new URLClassLoader(new URL[] { - new URL(url) - }); - thrown.expect(IllegalArgumentException.class); - thrown.expectMessage("Unable to convert url (" + url + ") to file."); - - DataflowRunner.detectClassPathResourcesToStage(classLoader); - } - - @Test public void testGcsStagingLocationInitialization() throws Exception { // Set temp location (required), and check that staging location is set. DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class); diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkContextOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkContextOptions.java index 98f7492..0a7995f 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkContextOptions.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkContextOptions.java @@ -54,7 +54,7 @@ public interface SparkContextOptions extends SparkPipelineOptions { List<JavaStreamingListener> getListeners(); void setListeners(List<JavaStreamingListener> listeners); - /** Returns an empty list, top avoid handling null. */ + /** Returns an empty list, to avoid handling null. */ class EmptyListenersList implements DefaultValueFactory<List<JavaStreamingListener>> { @Override public List<JavaStreamingListener> create(PipelineOptions options) { diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java index 2053b44..146f25b 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkPipelineOptions.java @@ -20,7 +20,6 @@ package org.apache.beam.runners.spark; import java.util.ArrayList; import java.util.List; - import org.apache.beam.sdk.options.ApplicationNameOptions; import org.apache.beam.sdk.options.Default; import org.apache.beam.sdk.options.DefaultValueFactory; @@ -28,8 +27,6 @@ import org.apache.beam.sdk.options.Description; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.options.StreamingOptions; - - /** * Spark runner {@link PipelineOptions} handles Spark execution-related configurations, * such as the master address, batch-interval, and other user-related knobs. @@ -104,12 +101,20 @@ public interface SparkPipelineOptions boolean getUsesProvidedSparkContext(); void setUsesProvidedSparkContext(boolean value); - @Description("Jars for spark context") - @Default.InstanceFactory(SparkContextOptions.EmptyPathList.class) - List<String> getJarsForSparkContext(); - void setJarsForSparkContext(List<String> jars); + /** + * List of local files to make available to workers. + * + * <p>Jars are placed on the worker's classpath. + * + * <p>The default value is the list of jars from the main program's classpath. + */ + @Description("Jar-Files to send to all workers and put on the classpath. " + + "The default value is all files from the classpath.") + @Default.InstanceFactory(EmptyPathList.class) + List<String> getFilesToStage(); + void setFilesToStage(List<String> value); - /** Returns an empty path list, top avoid handling null. */ + /** Returns an empty path list, to avoid handling null. */ class EmptyPathList implements DefaultValueFactory<List<String>> { @Override public List<String> create(PipelineOptions options) { diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java index 4a409cb..ccf8283 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/SparkRunner.java @@ -18,6 +18,8 @@ package org.apache.beam.runners.spark; +import static org.apache.beam.sdk.util.PipelineUtils.detectClassPathResourcesToStage; + import com.google.common.collect.Iterables; import java.util.Arrays; import java.util.Collection; @@ -121,6 +123,17 @@ public final class SparkRunner extends PipelineRunner<SparkPipelineResult> { public static SparkRunner fromOptions(PipelineOptions options) { SparkPipelineOptions sparkOptions = PipelineOptionsValidator.validate(SparkPipelineOptions.class, options); + + if (sparkOptions.getFilesToStage().isEmpty()) { + sparkOptions.setFilesToStage(detectClassPathResourcesToStage( + SparkRunner.class.getClassLoader())); + LOG.info("PipelineOptions.filesToStage was not specified. " + + "Defaulting to files from the classpath: will stage {} files. " + + "Enable logging at DEBUG level to see which files will be staged.", + sparkOptions.getFilesToStage().size()); + LOG.debug("Classpath elements: {}", sparkOptions.getFilesToStage()); + } + return new SparkRunner(sparkOptions); } diff --git a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java index 68c4093..d0b467a 100644 --- a/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java +++ b/runners/spark/src/main/java/org/apache/beam/runners/spark/translation/SparkContextFactory.java @@ -93,8 +93,8 @@ public final class SparkContextFactory { conf.setMaster(contextOptions.getSparkMaster()); } - if (contextOptions.getJarsForSparkContext().size() > 0) { - conf.setJars(contextOptions.getJarsForSparkContext().toArray(new String[0])); + if (contextOptions.getFilesToStage().size() > 0) { + conf.setJars(contextOptions.getFilesToStage().toArray(new String[0])); } conf.setAppName(contextOptions.getAppName()); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PipelineUtils.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PipelineUtils.java new file mode 100644 index 0000000..1614dca --- /dev/null +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/util/PipelineUtils.java @@ -0,0 +1,59 @@ +/* + * 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. + */ +package org.apache.beam.sdk.util; + +import java.io.File; +import java.net.URISyntaxException; +import java.net.URL; +import java.net.URLClassLoader; +import java.util.ArrayList; +import java.util.List; + +/** + * Utilities for working with Pipelines. + */ +public class PipelineUtils { + + /** + * Attempts to detect all the resources the class loader has access to. This does not recurse + * to class loader parents stopping it from pulling in resources from the system class loader. + * + * @param classLoader The URLClassLoader to use to detect resources to stage. + * @throws IllegalArgumentException If either the class loader is not a URLClassLoader or one + * of the resources the class loader exposes is not a file resource. + * @return A list of absolute paths to the resources the class loader uses. + */ + public static List<String> detectClassPathResourcesToStage(ClassLoader classLoader) { + if (!(classLoader instanceof URLClassLoader)) { + String message = String.format("Unable to use ClassLoader to detect classpath elements. " + + "Current ClassLoader is %s, only URLClassLoaders are supported.", classLoader); + throw new IllegalArgumentException(message); + } + + List<String> files = new ArrayList<>(); + for (URL url : ((URLClassLoader) classLoader).getURLs()) { + try { + files.add(new File(url.toURI()).getAbsolutePath()); + } catch (IllegalArgumentException | URISyntaxException e) { + String message = String.format("Unable to convert url (%s) to file.", url); + throw new IllegalArgumentException(message, e); + } + } + return files; + } +} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PipelineUtilsTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PipelineUtilsTest.java new file mode 100644 index 0000000..4d03615 --- /dev/null +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/PipelineUtilsTest.java @@ -0,0 +1,76 @@ +/* + * 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. + */ +package org.apache.beam.sdk.util; + +import static org.junit.Assert.assertEquals; + +import com.google.common.collect.ImmutableList; +import java.io.File; +import java.net.URL; +import java.net.URLClassLoader; +import org.junit.Rule; +import org.junit.Test; +import org.junit.rules.ExpectedException; +import org.junit.rules.TemporaryFolder; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; +import org.mockito.Mockito; + +/** + * Tests for PipelineUtils. + */ +@RunWith(JUnit4.class) +public class PipelineUtilsTest { + + @Rule public transient TemporaryFolder tmpFolder = new TemporaryFolder(); + @Rule public transient ExpectedException thrown = ExpectedException.none(); + + @Test + public void detectClassPathResourceWithFileResources() throws Exception { + File file = tmpFolder.newFile("file"); + File file2 = tmpFolder.newFile("file2"); + URLClassLoader classLoader = new URLClassLoader(new URL[] { + file.toURI().toURL(), + file2.toURI().toURL() + }); + + assertEquals(ImmutableList.of(file.getAbsolutePath(), file2.getAbsolutePath()), + PipelineUtils.detectClassPathResourcesToStage(classLoader)); + } + + @Test + public void detectClassPathResourcesWithUnsupportedClassLoader() { + ClassLoader mockClassLoader = Mockito.mock(ClassLoader.class); + thrown.expect(IllegalArgumentException.class); + thrown.expectMessage("Unable to use ClassLoader to detect classpath elements."); + + PipelineUtils.detectClassPathResourcesToStage(mockClassLoader); + } + + @Test + public void detectClassPathResourceWithNonFileResources() throws Exception { + String url = "http://www.google.com/all-the-secrets.jar"; + URLClassLoader classLoader = new URLClassLoader(new URL[] { + new URL(url) + }); + thrown.expect(IllegalArgumentException.class); + thrown.expectMessage("Unable to convert url (" + url + ") to file."); + + PipelineUtils.detectClassPathResourcesToStage(classLoader); + } +} -- To stop receiving notification emails like this one, please contact "commits@beam.apache.org" <commits@beam.apache.org>.