Replace GcsPath by IOChannelFactory in WordCount.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/60583307 Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/60583307 Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/60583307 Branch: refs/heads/python-sdk Commit: 605833071a7034aa3b723776a0f9e24330f64c8b Parents: e90a1b9 Author: Pei He <[email protected]> Authored: Mon Jun 13 16:58:01 2016 -0700 Committer: Davor Bonaci <[email protected]> Committed: Mon Jun 20 15:14:29 2016 -0700 ---------------------------------------------------------------------- .../java/org/apache/beam/examples/WordCount.java | 19 +++++++++++++++---- 1 file changed, 15 insertions(+), 4 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/60583307/examples/java/src/main/java/org/apache/beam/examples/WordCount.java ---------------------------------------------------------------------- diff --git a/examples/java/src/main/java/org/apache/beam/examples/WordCount.java b/examples/java/src/main/java/org/apache/beam/examples/WordCount.java index f5611d3..03ed5d7 100644 --- a/examples/java/src/main/java/org/apache/beam/examples/WordCount.java +++ b/examples/java/src/main/java/org/apache/beam/examples/WordCount.java @@ -32,10 +32,15 @@ import org.apache.beam.sdk.transforms.PTransform; import org.apache.beam.sdk.transforms.ParDo; import org.apache.beam.sdk.transforms.SimpleFunction; import org.apache.beam.sdk.transforms.Sum; -import org.apache.beam.sdk.util.gcsfs.GcsPath; +import org.apache.beam.sdk.util.IOChannelFactory; +import org.apache.beam.sdk.util.IOChannelUtils; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; +import com.google.common.base.Strings; + +import java.io.IOException; + /** * An example that counts words in Shakespeare and includes Beam best practices. * @@ -176,9 +181,15 @@ public class WordCount { public static class OutputFactory implements DefaultValueFactory<String> { @Override public String create(PipelineOptions options) { - if (options.getTempLocation() != null) { - return GcsPath.fromUri(options.getTempLocation()) - .resolve("counts.txt").toString(); + String tempLocation = options.getTempLocation(); + if (!Strings.isNullOrEmpty(tempLocation)) { + try { + IOChannelFactory factory = IOChannelUtils.getFactory(tempLocation); + return factory.resolve(tempLocation, "counts.txt"); + } catch (IOException e) { + throw new RuntimeException( + String.format("Failed to resolve temp location: %s", tempLocation)); + } } else { throw new IllegalArgumentException("Must specify --output or --tempLocation"); }
