You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by lc...@apache.org on 2016/04/07 20:21:11 UTC
[01/12] incubator-beam git commit: [BEAM-151] Break out Dataflow
runner dependency to separate test file
Repository: incubator-beam
Updated Branches:
refs/heads/master 7f0a2f78a -> 6467afb74
[BEAM-151] Break out Dataflow runner dependency to separate test file
This allows for moving the Dataflow specific portion of the testing to the Dataflow runner maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/05f7aaa4
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/05f7aaa4
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/05f7aaa4
Branch: refs/heads/master
Commit: 05f7aaa491e714d91f8a4740cbea48638f2d40ee
Parents: a18ab87
Author: Luke Cwik <lc...@google.com>
Authored: Fri Mar 25 16:19:54 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:18:32 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/DataflowViewTest.java | 205 +++++++++++++++++++
.../cloud/dataflow/sdk/transforms/ViewTest.java | 126 ------------
2 files changed, 205 insertions(+), 126 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/05f7aaa4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
new file mode 100644
index 0000000..b67bfcf
--- /dev/null
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
@@ -0,0 +1,205 @@
+/*
+ * Copyright (C) 2015 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.transforms;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.windowing.FixedWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.InvalidWindows;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
+import com.google.cloud.dataflow.sdk.util.NoopPathValidator;
+import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PBegin;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+import com.google.cloud.dataflow.sdk.values.PCollectionView;
+import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
+
+import org.hamcrest.Matchers;
+import org.joda.time.Duration;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.internal.matchers.ThrowableMessageMatcher;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+/** Tests for {@link View} for a {@link DataflowPipelineRunner}. */
+@RunWith(JUnit4.class)
+public class DataflowViewTest {
+ @Rule
+ public transient ExpectedException thrown = ExpectedException.none();
+
+ private Pipeline createTestBatchRunner() {
+ DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ options.setRunner(DataflowPipelineRunner.class);
+ options.setProject("someproject");
+ options.setStagingLocation("gs://staging");
+ options.setPathValidatorClass(NoopPathValidator.class);
+ options.setDataflowClient(null);
+ return Pipeline.create(options);
+ }
+
+ private Pipeline createTestStreamingRunner() {
+ DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ options.setRunner(DataflowPipelineRunner.class);
+ options.setStreaming(true);
+ options.setProject("someproject");
+ options.setStagingLocation("gs://staging");
+ options.setPathValidatorClass(NoopPathValidator.class);
+ options.setDataflowClient(null);
+ return Pipeline.create(options);
+ }
+
+ private void testViewUnbounded(
+ Pipeline pipeline,
+ PTransform<PCollection<KV<String, Integer>>, ? extends PCollectionView<?>> view) {
+ thrown.expect(IllegalStateException.class);
+ thrown.expectMessage("Unable to create a side-input view from input");
+ thrown.expectCause(
+ ThrowableMessageMatcher.hasMessage(Matchers.containsString("non-bounded PCollection")));
+ pipeline
+ .apply(
+ new PTransform<PBegin, PCollection<KV<String, Integer>>>() {
+ @Override
+ public PCollection<KV<String, Integer>> apply(PBegin input) {
+ return PCollection.<KV<String, Integer>>createPrimitiveOutputInternal(
+ input.getPipeline(),
+ WindowingStrategy.globalDefault(),
+ PCollection.IsBounded.UNBOUNDED)
+ .setTypeDescriptorInternal(new TypeDescriptor<KV<String, Integer>>() {});
+ }
+ })
+ .apply(view);
+ }
+
+ private void testViewNonmerging(
+ Pipeline pipeline,
+ PTransform<PCollection<KV<String, Integer>>, ? extends PCollectionView<?>> view) {
+ thrown.expect(IllegalStateException.class);
+ thrown.expectMessage("Unable to create a side-input view from input");
+ thrown.expectCause(
+ ThrowableMessageMatcher.hasMessage(Matchers.containsString("Consumed by GroupByKey")));
+ pipeline.apply(Create.<KV<String, Integer>>of(KV.of("hello", 5)))
+ .apply(Window.<KV<String, Integer>>into(new InvalidWindows<>(
+ "Consumed by GroupByKey", FixedWindows.of(Duration.standardHours(1)))))
+ .apply(view);
+ }
+
+ @Test
+ public void testViewUnboundedAsSingletonBatch() {
+ testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asSingleton());
+ }
+
+ @Test
+ public void testViewUnboundedAsSingletonStreaming() {
+ testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asSingleton());
+ }
+
+ @Test
+ public void testViewUnboundedAsIterableBatch() {
+ testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asIterable());
+ }
+
+ @Test
+ public void testViewUnboundedAsIterableStreaming() {
+ testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asIterable());
+ }
+
+ @Test
+ public void testViewUnboundedAsListBatch() {
+ testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asList());
+ }
+
+ @Test
+ public void testViewUnboundedAsListStreaming() {
+ testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asList());
+ }
+
+ @Test
+ public void testViewUnboundedAsMapBatch() {
+ testViewUnbounded(createTestBatchRunner(), View.<String, Integer>asMap());
+ }
+
+ @Test
+ public void testViewUnboundedAsMapStreaming() {
+ testViewUnbounded(createTestStreamingRunner(), View.<String, Integer>asMap());
+ }
+
+ @Test
+ public void testViewUnboundedAsMultimapBatch() {
+ testViewUnbounded(createTestBatchRunner(), View.<String, Integer>asMultimap());
+ }
+
+ @Test
+ public void testViewUnboundedAsMultimapStreaming() {
+ testViewUnbounded(createTestStreamingRunner(), View.<String, Integer>asMultimap());
+ }
+
+ @Test
+ public void testViewNonmergingAsSingletonBatch() {
+ testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asSingleton());
+ }
+
+ @Test
+ public void testViewNonmergingAsSingletonStreaming() {
+ testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asSingleton());
+ }
+
+ @Test
+ public void testViewNonmergingAsIterableBatch() {
+ testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asIterable());
+ }
+
+ @Test
+ public void testViewNonmergingAsIterableStreaming() {
+ testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asIterable());
+ }
+
+ @Test
+ public void testViewNonmergingAsListBatch() {
+ testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asList());
+ }
+
+ @Test
+ public void testViewNonmergingAsListStreaming() {
+ testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asList());
+ }
+
+ @Test
+ public void testViewNonmergingAsMapBatch() {
+ testViewNonmerging(createTestBatchRunner(), View.<String, Integer>asMap());
+ }
+
+ @Test
+ public void testViewNonmergingAsMapStreaming() {
+ testViewNonmerging(createTestStreamingRunner(), View.<String, Integer>asMap());
+ }
+
+ @Test
+ public void testViewNonmergingAsMultimapBatch() {
+ testViewNonmerging(createTestBatchRunner(), View.<String, Integer>asMultimap());
+ }
+
+ @Test
+ public void testViewNonmergingAsMultimapStreaming() {
+ testViewNonmerging(createTestStreamingRunner(), View.<String, Integer>asMultimap());
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/05f7aaa4/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/ViewTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/ViewTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/ViewTest.java
index 64ec176..69a4d2e 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/ViewTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/ViewTest.java
@@ -33,10 +33,8 @@ import com.google.cloud.dataflow.sdk.coders.NullableCoder;
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
import com.google.cloud.dataflow.sdk.coders.VarIntCoder;
import com.google.cloud.dataflow.sdk.coders.VoidCoder;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.DirectPipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
@@ -45,7 +43,6 @@ import com.google.cloud.dataflow.sdk.transforms.windowing.FixedWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.InvalidWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
-import com.google.cloud.dataflow.sdk.util.NoopPathValidator;
import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PBegin;
@@ -1333,27 +1330,6 @@ public class ViewTest implements Serializable {
assertEquals("View.AsMultimap", View.<String, Integer>asMultimap().getName());
}
- private Pipeline createTestBatchRunner() {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
- options.setRunner(DataflowPipelineRunner.class);
- options.setProject("someproject");
- options.setStagingLocation("gs://staging");
- options.setPathValidatorClass(NoopPathValidator.class);
- options.setDataflowClient(null);
- return Pipeline.create(options);
- }
-
- private Pipeline createTestStreamingRunner() {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
- options.setRunner(DataflowPipelineRunner.class);
- options.setStreaming(true);
- options.setProject("someproject");
- options.setStagingLocation("gs://staging");
- options.setPathValidatorClass(NoopPathValidator.class);
- options.setDataflowClient(null);
- return Pipeline.create(options);
- }
-
private Pipeline createTestDirectRunner() {
DirectPipelineOptions options = PipelineOptionsFactory.as(DirectPipelineOptions.class);
options.setRunner(DirectPipelineRunner.class);
@@ -1396,152 +1372,50 @@ public class ViewTest implements Serializable {
}
@Test
- public void testViewUnboundedAsSingletonBatch() {
- testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asSingleton());
- }
-
- @Test
- public void testViewUnboundedAsSingletonStreaming() {
- testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asSingleton());
- }
-
- @Test
public void testViewUnboundedAsSingletonDirect() {
testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asSingleton());
}
@Test
- public void testViewUnboundedAsIterableBatch() {
- testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asIterable());
- }
-
- @Test
- public void testViewUnboundedAsIterableStreaming() {
- testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asIterable());
- }
-
- @Test
public void testViewUnboundedAsIterableDirect() {
testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asIterable());
}
@Test
- public void testViewUnboundedAsListBatch() {
- testViewUnbounded(createTestBatchRunner(), View.<KV<String, Integer>>asList());
- }
-
- @Test
- public void testViewUnboundedAsListStreaming() {
- testViewUnbounded(createTestStreamingRunner(), View.<KV<String, Integer>>asList());
- }
-
- @Test
public void testViewUnboundedAsListDirect() {
testViewUnbounded(createTestDirectRunner(), View.<KV<String, Integer>>asList());
}
@Test
- public void testViewUnboundedAsMapBatch() {
- testViewUnbounded(createTestBatchRunner(), View.<String, Integer>asMap());
- }
-
- @Test
- public void testViewUnboundedAsMapStreaming() {
- testViewUnbounded(createTestStreamingRunner(), View.<String, Integer>asMap());
- }
-
- @Test
public void testViewUnboundedAsMapDirect() {
testViewUnbounded(createTestDirectRunner(), View.<String, Integer>asMap());
}
-
- @Test
- public void testViewUnboundedAsMultimapBatch() {
- testViewUnbounded(createTestBatchRunner(), View.<String, Integer>asMultimap());
- }
-
- @Test
- public void testViewUnboundedAsMultimapStreaming() {
- testViewUnbounded(createTestStreamingRunner(), View.<String, Integer>asMultimap());
- }
-
@Test
public void testViewUnboundedAsMultimapDirect() {
testViewUnbounded(createTestDirectRunner(), View.<String, Integer>asMultimap());
}
@Test
- public void testViewNonmergingAsSingletonBatch() {
- testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asSingleton());
- }
-
- @Test
- public void testViewNonmergingAsSingletonStreaming() {
- testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asSingleton());
- }
-
- @Test
public void testViewNonmergingAsSingletonDirect() {
testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asSingleton());
}
@Test
- public void testViewNonmergingAsIterableBatch() {
- testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asIterable());
- }
-
- @Test
- public void testViewNonmergingAsIterableStreaming() {
- testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asIterable());
- }
-
- @Test
public void testViewNonmergingAsIterableDirect() {
testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asIterable());
}
@Test
- public void testViewNonmergingAsListBatch() {
- testViewNonmerging(createTestBatchRunner(), View.<KV<String, Integer>>asList());
- }
-
- @Test
- public void testViewNonmergingAsListStreaming() {
- testViewNonmerging(createTestStreamingRunner(), View.<KV<String, Integer>>asList());
- }
-
- @Test
public void testViewNonmergingAsListDirect() {
testViewNonmerging(createTestDirectRunner(), View.<KV<String, Integer>>asList());
}
@Test
- public void testViewNonmergingAsMapBatch() {
- testViewNonmerging(createTestBatchRunner(), View.<String, Integer>asMap());
- }
-
- @Test
- public void testViewNonmergingAsMapStreaming() {
- testViewNonmerging(createTestStreamingRunner(), View.<String, Integer>asMap());
- }
-
- @Test
public void testViewNonmergingAsMapDirect() {
testViewNonmerging(createTestDirectRunner(), View.<String, Integer>asMap());
}
-
- @Test
- public void testViewNonmergingAsMultimapBatch() {
- testViewNonmerging(createTestBatchRunner(), View.<String, Integer>asMultimap());
- }
-
- @Test
- public void testViewNonmergingAsMultimapStreaming() {
- testViewNonmerging(createTestStreamingRunner(), View.<String, Integer>asMultimap());
- }
-
@Test
public void testViewNonmergingAsMultimapDirect() {
testViewNonmerging(createTestDirectRunner(), View.<String, Integer>asMultimap());
[05/12] incubator-beam git commit: [BEAM-151] Remove dependence on
num workers in DatastoreIO
Posted by lc...@apache.org.
[BEAM-151] Remove dependence on num workers in DatastoreIO
This prevents moving DataflowPipelineOptions into a separate maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/551d0814
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/551d0814
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/551d0814
Branch: refs/heads/master
Commit: 551d081481450e750a9a95b5902a540d0662db85
Parents: d0db477
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 13:11:28 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../com/google/cloud/dataflow/sdk/io/DatastoreIO.java | 14 +-------------
1 file changed, 1 insertion(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/551d0814/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/DatastoreIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/DatastoreIO.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/DatastoreIO.java
index 9128585..8d85ab3 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/DatastoreIO.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/DatastoreIO.java
@@ -54,7 +54,6 @@ import com.google.cloud.dataflow.sdk.coders.EntityCoder;
import com.google.cloud.dataflow.sdk.coders.SerializableCoder;
import com.google.cloud.dataflow.sdk.io.Sink.WriteOperation;
import com.google.cloud.dataflow.sdk.io.Sink.Writer;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineWorkerPoolOptions;
import com.google.cloud.dataflow.sdk.options.GcpOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.util.AttemptBoundedExponentialBackOff;
@@ -307,18 +306,7 @@ public class DatastoreIO {
numSplits = Math.round(((double) getEstimatedSizeBytes(options)) / desiredBundleSizeBytes);
} catch (Exception e) {
// Fallback in case estimated size is unavailable. TODO: fix this, it's horrible.
-
- // 1. Try Dataflow's numWorkers, which will be 0 for other workers.
- DataflowPipelineWorkerPoolOptions poolOptions =
- options.as(DataflowPipelineWorkerPoolOptions.class);
- if (poolOptions.getNumWorkers() > 0) {
- LOG.warn("Estimated size of unavailable, using the number of workers {}",
- poolOptions.getNumWorkers(), e);
- numSplits = poolOptions.getNumWorkers();
- } else {
- // 2. Default to 12 in the unknown case.
- numSplits = 12;
- }
+ numSplits = 12;
}
// If the desiredBundleSize or number of workers results in 1 split, simply return
[10/12] incubator-beam git commit: [BEAM-151] Remove dependence on
DataflowPipelineOptions
Posted by lc...@apache.org.
[BEAM-151] Remove dependence on DataflowPipelineOptions
This prevents moving DataflowPipelineOptions to its own runner specifc maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/e0d4f85d
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/e0d4f85d
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/e0d4f85d
Branch: refs/heads/master
Commit: e0d4f85d4be3f55dc655258ffaa6eef213894df0
Parents: 49d6594
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 14:16:34 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:50 2016 -0700
----------------------------------------------------------------------
.../cloud/dataflow/sdk/util/Transport.java | 16 ++--------
.../dataflow/sdk/testing/TestPipelineTest.java | 32 +++++---------------
2 files changed, 10 insertions(+), 38 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e0d4f85d/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
index 5888822..d5cdb24 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
@@ -27,7 +27,6 @@ import com.google.api.services.bigquery.Bigquery;
import com.google.api.services.pubsub.Pubsub;
import com.google.api.services.storage.Storage;
import com.google.cloud.dataflow.sdk.options.BigQueryOptions;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineDebugOptions;
import com.google.cloud.dataflow.sdk.options.GcsOptions;
import com.google.cloud.dataflow.sdk.options.PubsubOptions;
import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer;
@@ -90,10 +89,7 @@ public class Transport {
}
/**
- * Returns a BigQuery client builder.
- *
- * <p>Note: this client's endpoint is <b>not</b> modified by the
- * {@link DataflowPipelineDebugOptions#getApiRootUrl()} option.
+ * Returns a BigQuery client builder using the specified {@link BigQueryOptions}.
*/
public static Bigquery.Builder
newBigQueryClient(BigQueryOptions options) {
@@ -107,10 +103,7 @@ public class Transport {
}
/**
- * Returns a Pubsub client builder.
- *
- * <p>Note: this client's endpoint is <b>not</b> modified by the
- * {@link DataflowPipelineDebugOptions#getApiRootUrl()} option.
+ * Returns a Pubsub client builder using the specified {@link PubsubOptions}.
*/
public static Pubsub.Builder
newPubsubClient(PubsubOptions options) {
@@ -125,10 +118,7 @@ public class Transport {
}
/**
- * Returns a Cloud Storage client builder.
- *
- * <p>Note: this client's endpoint is <b>not</b> modified by the
- * {@link DataflowPipelineDebugOptions#getApiRootUrl()} option.
+ * Returns a Cloud Storage client builder using the specified {@link GcsOptions}.
*/
public static Storage.Builder
newStorageClient(GcsOptions options) {
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/e0d4f85d/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/TestPipelineTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/TestPipelineTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/TestPipelineTest.java
index 6f119da..d6a2b63 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/TestPipelineTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/testing/TestPipelineTest.java
@@ -23,10 +23,9 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertThat;
import com.google.cloud.dataflow.sdk.options.ApplicationNameOptions;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.GcpOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.fasterxml.jackson.databind.ObjectMapper;
@@ -50,31 +49,14 @@ public class TestPipelineTest {
public void testCreationOfPipelineOptions() throws Exception {
ObjectMapper mapper = new ObjectMapper();
String stringOptions = mapper.writeValueAsString(new String[]{
- "--runner=DataflowPipelineRunner",
- "--project=testProject",
- "--apiRootUrl=testApiRootUrl",
- "--dataflowEndpoint=testDataflowEndpoint",
- "--tempLocation=testTempLocation",
- "--serviceAccountName=testServiceAccountName",
- "--serviceAccountKeyfile=testServiceAccountKeyfile",
- "--zone=testZone",
- "--numWorkers=1",
- "--diskSizeGb=2"
+ "--runner=DirectPipelineRunner",
+ "--project=testProject"
});
System.getProperties().put("dataflowOptions", stringOptions);
- DataflowPipelineOptions options =
- TestPipeline.testingPipelineOptions().as(DataflowPipelineOptions.class);
- assertEquals(DataflowPipelineRunner.class, options.getRunner());
- assertThat(options.getJobName(), startsWith("testpipelinetest0testcreationofpipelineoptions-"));
- assertEquals("testProject", options.as(GcpOptions.class).getProject());
- assertEquals("testApiRootUrl", options.getApiRootUrl());
- assertEquals("testDataflowEndpoint", options.getDataflowEndpoint());
- assertEquals("testTempLocation", options.getTempLocation());
- assertEquals("testServiceAccountName", options.getServiceAccountName());
- assertEquals(
- "testServiceAccountKeyfile", options.as(GcpOptions.class).getServiceAccountKeyfile());
- assertEquals("testZone", options.getZone());
- assertEquals(2, options.getDiskSizeGb());
+ GcpOptions options =
+ TestPipeline.testingPipelineOptions().as(GcpOptions.class);
+ assertEquals(DirectPipelineRunner.class, options.getRunner());
+ assertEquals(options.getProject(), "testProject");
}
@Test
[11/12] incubator-beam git commit: [BEAM-151] Update ASF license at
top of new files.
Posted by lc...@apache.org.
[BEAM-151] Update ASF license at top of new files.
Also fix minor comment.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/ac1dd437
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/ac1dd437
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/ac1dd437
Branch: refs/heads/master
Commit: ac1dd437a38a2e46eccc35e1601f8858eba72281
Parents: e0d4f85
Author: Luke Cwik <lc...@google.com>
Authored: Thu Mar 31 09:56:48 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:50 2016 -0700
----------------------------------------------------------------------
.../dataflow/sdk/options/PubsubOptions.java | 25 +++++++++---------
.../dataflow/sdk/util/DataflowTransport.java | 27 ++++++++++----------
.../dataflow/sdk/io/DataflowTextIOTest.java | 23 +++++++++--------
.../sdk/transforms/DataflowGroupByKeyTest.java | 23 +++++++++--------
.../sdk/transforms/DataflowViewTest.java | 23 +++++++++--------
5 files changed, 63 insertions(+), 58 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac1dd437/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
index 51cf62c..deb19e9 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
@@ -1,23 +1,24 @@
/*
- * Copyright (C) 2016 Google Inc.
+ * 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
*
- * Licensed 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
+ * 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.
+ * 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 com.google.cloud.dataflow.sdk.options;
/**
- * Properties that can be set when using Pubsub with the Dataflow SDK.
+ * Properties that can be set when using Pubsub with the Beam SDK.
*/
@Description("Options that are used to configure BigQuery. See "
+ "https://cloud.google.com/bigquery/what-is-bigquery for details on BigQuery.")
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac1dd437/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
index 8de358c..18e6654 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
@@ -1,19 +1,20 @@
-/*******************************************************************************
- * Copyright (C) 2015 Google Inc.
+/*
+ * 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
*
- * Licensed 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
+ * 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.
- ******************************************************************************/
-
+ * 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 com.google.cloud.dataflow.sdk.util;
import static com.google.cloud.dataflow.sdk.util.Transport.getJsonFactory;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac1dd437/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
index f6d68ec..1bd8a85 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
@@ -1,19 +1,20 @@
/*
- * Copyright (C) 2016 Google Inc.
+ * 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
*
- * Licensed 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
+ * 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.
+ * 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 com.google.cloud.dataflow.sdk.io;
import com.google.cloud.dataflow.sdk.Pipeline;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac1dd437/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
index b05e7a2..b0b011d 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
@@ -1,19 +1,20 @@
/*
- * Copyright (C) 2016 Google Inc.
+ * 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
*
- * Licensed 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
+ * 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.
+ * 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 com.google.cloud.dataflow.sdk.transforms;
import com.google.cloud.dataflow.sdk.Pipeline;
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/ac1dd437/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
index b67bfcf..c2a4273 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowViewTest.java
@@ -1,19 +1,20 @@
/*
- * Copyright (C) 2015 Google Inc.
+ * 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
*
- * Licensed 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
+ * 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.
+ * 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 com.google.cloud.dataflow.sdk.transforms;
import com.google.cloud.dataflow.sdk.Pipeline;
[06/12] incubator-beam git commit: [BEAM-151] Fixup of comment in
test file
Posted by lc...@apache.org.
[BEAM-151] Fixup of comment in test file
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a502baa9
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a502baa9
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a502baa9
Branch: refs/heads/master
Commit: a502baa9171f610a9988774fa654e27cf5fcba79
Parents: 22fd120
Author: Luke Cwik <lc...@google.com>
Authored: Fri Mar 25 16:29:48 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java | 2 +-
1 file changed, 1 insertion(+), 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a502baa9/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
index 8c0d6d2..f6d68ec 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
@@ -1,5 +1,5 @@
/*
- * Copyright (C) 2015 Google Inc.
+ * Copyright (C) 2016 Google Inc.
*
* Licensed 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
[12/12] incubator-beam git commit: [BEAM-151] This closes #87
Posted by lc...@apache.org.
[BEAM-151] This closes #87
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/6467afb7
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/6467afb7
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/6467afb7
Branch: refs/heads/master
Commit: 6467afb74b1b37cccc3f4210a08e75e678612010
Parents: 7f0a2f7 ac1dd43
Author: Luke Cwik <lc...@google.com>
Authored: Thu Apr 7 11:20:55 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:20:55 2016 -0700
----------------------------------------------------------------------
.../cloud/dataflow/sdk/io/DatastoreIO.java | 14 +-
.../google/cloud/dataflow/sdk/io/PubsubIO.java | 14 +-
.../google/cloud/dataflow/sdk/io/TextIO.java | 10 +-
.../options/DataflowPipelineDebugOptions.java | 13 +-
.../sdk/options/DataflowPipelineOptions.java | 2 +-
.../dataflow/sdk/options/PubsubOptions.java | 36 ++++
.../sdk/runners/DataflowPipelineRunner.java | 6 +-
.../dataflow/sdk/util/DataflowTransport.java | 112 ++++++++++
.../cloud/dataflow/sdk/util/Transport.java | 65 +-----
.../dataflow/sdk/io/DataflowTextIOTest.java | 118 +++++++++++
.../cloud/dataflow/sdk/io/DatastoreIOTest.java | 66 +-----
.../cloud/dataflow/sdk/io/TextIOTest.java | 79 +------
.../sdk/options/GoogleApiDebugOptionsTest.java | 93 +++++----
.../sdk/options/PipelineOptionsFactoryTest.java | 11 +-
.../dataflow/sdk/testing/TestPipelineTest.java | 32 +--
.../sdk/transforms/DataflowGroupByKeyTest.java | 111 ++++++++++
.../sdk/transforms/DataflowViewTest.java | 206 +++++++++++++++++++
.../dataflow/sdk/transforms/GroupByKeyTest.java | 62 ------
.../cloud/dataflow/sdk/transforms/ViewTest.java | 126 ------------
19 files changed, 678 insertions(+), 498 deletions(-)
----------------------------------------------------------------------
[08/12] incubator-beam git commit: [BEAM-151] Remove references to
DataflowPipelineOptions
Posted by lc...@apache.org.
[BEAM-151] Remove references to DataflowPipelineOptions
Removed DatastoreIO dependence on numWorkers from DataflowPipelineWorkerHarnessOptions.
This prevented moving DataflowPipelineOptions to a separate maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c8f09a08
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c8f09a08
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c8f09a08
Branch: refs/heads/master
Commit: c8f09a08dad8b90643ae5bf55d195681192fcbdc
Parents: 551d081
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 13:19:53 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../google/cloud/dataflow/sdk/io/PubsubIO.java | 8 +--
.../google/cloud/dataflow/sdk/io/TextIO.java | 10 ++-
.../cloud/dataflow/sdk/io/DatastoreIOTest.java | 66 ++++----------------
3 files changed, 18 insertions(+), 66 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8f09a08/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
index 68615c5..deed9ab 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
@@ -34,7 +34,6 @@ import com.google.cloud.dataflow.sdk.coders.Coder;
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
import com.google.cloud.dataflow.sdk.coders.VoidCoder;
import com.google.cloud.dataflow.sdk.options.PubsubOptions;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
import com.google.cloud.dataflow.sdk.transforms.Create;
@@ -399,10 +398,9 @@ public class PubsubIO {
* the stream.
*
* <p>When running with a {@link PipelineRunner} that only supports bounded
- * {@link PCollection PCollections} (such as {@link DirectPipelineRunner} or
- * {@link DataflowPipelineRunner} without {@code --streaming}), only a bounded portion of the
- * input Pub/Sub stream can be processed. As such, either {@link Bound#maxNumRecords(int)} or
- * {@link Bound#maxReadTime(Duration)} must be set.
+ * {@link PCollection PCollections} (such as {@link DirectPipelineRunner}),
+ * only a bounded portion of the input Pub/Sub stream can be processed. As such, either
+ * {@link Bound#maxNumRecords(int)} or {@link Bound#maxReadTime(Duration)} must be set.
*/
public static class Read {
/**
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8f09a08/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java
index 0980d48..ed25926 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/TextIO.java
@@ -25,7 +25,6 @@ import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
import com.google.cloud.dataflow.sdk.coders.VoidCoder;
import com.google.cloud.dataflow.sdk.io.Read.Bounded;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
import com.google.cloud.dataflow.sdk.util.IOChannelUtils;
@@ -113,11 +112,10 @@ import javax.annotation.Nullable;
* <h3>Permissions</h3>
* <p>When run using the {@link DirectPipelineRunner}, your pipeline can read and write text files
* on your local drive and remote text files on Google Cloud Storage that you have access to using
- * your {@code gcloud} credentials. When running in the Dataflow service using
- * {@link DataflowPipelineRunner}, the pipeline can only read and write files from GCS. For more
- * information about permissions, see the Cloud Dataflow documentation on
- * <a href="https://cloud.google.com/dataflow/security-and-permissions">Security and
- * Permissions</a>.
+ * your {@code gcloud} credentials. When running in the Dataflow service, the pipeline can only
+ * read and write files from GCS. For more information about permissions, see the Cloud Dataflow
+ * documentation on <a href="https://cloud.google.com/dataflow/security-and-permissions">Security
+ * and Permissions</a>.
*/
public class TextIO {
/** The default coder, which returns each line of the input file as a string. */
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8f09a08/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DatastoreIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DatastoreIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DatastoreIOTest.java
index 4196740..dcdb229 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DatastoreIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DatastoreIOTest.java
@@ -51,7 +51,7 @@ import com.google.api.services.datastore.client.DatastoreHelper;
import com.google.api.services.datastore.client.QuerySplitter;
import com.google.cloud.dataflow.sdk.io.DatastoreIO.DatastoreReader;
import com.google.cloud.dataflow.sdk.io.DatastoreIO.DatastoreWriter;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.GcpOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.testing.ExpectedLogs;
@@ -69,14 +69,11 @@ import org.mockito.MockitoAnnotations;
import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
-import java.io.IOException;
import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.NoSuchElementException;
-import javax.annotation.Nullable;
-
/**
* Tests for {@link DatastoreIO}.
*/
@@ -112,12 +109,9 @@ public class DatastoreIOTest {
/**
* Helper function to create a test {@code DataflowPipelineOptions}.
*/
- static final DataflowPipelineOptions testPipelineOptions(@Nullable Integer numWorkers) {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ static final GcpOptions testPipelineOptions() {
+ GcpOptions options = PipelineOptionsFactory.as(GcpOptions.class);
options.setGcpCredential(new TestCredential());
- if (numWorkers != null) {
- options.setNumWorkers(numWorkers);
- }
return options;
}
@@ -217,13 +211,13 @@ public class DatastoreIOTest {
thrown.expect(NullPointerException.class);
thrown.expectMessage("Dataset");
- sink.validate(testPipelineOptions(null));
+ sink.validate(testPipelineOptions());
}
@Test
public void testSinkValidationSucceedsWithDataset() throws Exception {
DatastoreIO.Sink sink = DatastoreIO.sink().withDataset(DATASET);
- sink.validate(testPipelineOptions(null));
+ sink.validate(testPipelineOptions());
}
@Test
@@ -254,7 +248,7 @@ public class DatastoreIOTest {
.withMockSplitter(splitter)
.withMockEstimateSizeBytes(8 * 1024L);
- List<DatastoreIO.Source> bundles = io.splitIntoBundles(1024, testPipelineOptions(null));
+ List<DatastoreIO.Source> bundles = io.splitIntoBundles(1024, testPipelineOptions());
assertEquals(8, bundles.size());
for (int i = 0; i < 8; ++i) {
DatastoreIO.Source bundle = bundles.get(i);
@@ -274,7 +268,7 @@ public class DatastoreIOTest {
.withMockSplitter(splitter)
.withMockEstimateSizeBytes(8 * 1024L);
- io.splitIntoBundles(1024, testPipelineOptions(null));
+ io.splitIntoBundles(1024, testPipelineOptions());
PartitionId partition = PartitionId.newBuilder().setNamespace(NAMESPACE).build();
verify(splitter).getSplits(eq(QUERY), eq(partition), eq(8), any(Datastore.class));
@@ -300,7 +294,7 @@ public class DatastoreIOTest {
.withMockSplitter(splitter)
.withMockEstimateSizeBytes(0L);
- List<DatastoreIO.Source> bundles = io.splitIntoBundles(1024, testPipelineOptions(null));
+ List<DatastoreIO.Source> bundles = io.splitIntoBundles(1024, testPipelineOptions());
assertEquals(1, bundles.size());
verify(splitter, never())
.getSplits(any(Query.class), any(PartitionId.class), eq(1), any(Datastore.class));
@@ -328,7 +322,7 @@ public class DatastoreIOTest {
initialSource
.withQuery(query)
.withMockSplitter(splitter)
- .splitIntoBundles(1024, testPipelineOptions(null));
+ .splitIntoBundles(1024, testPipelineOptions());
assertEquals(1, bundles.size());
assertEquals(query, bundles.get(0).getQuery());
@@ -356,7 +350,7 @@ public class DatastoreIOTest {
.withQuery(query)
.withMockSplitter(splitter)
.withMockEstimateSizeBytes(10240L)
- .splitIntoBundles(1024, testPipelineOptions(null));
+ .splitIntoBundles(1024, testPipelineOptions());
assertEquals(1, bundles.size());
assertEquals(query, bundles.get(0).getQuery());
@@ -371,44 +365,6 @@ public class DatastoreIOTest {
KindExpression mykind = KindExpression.newBuilder().setName("mykind").build();
Query query = Query.newBuilder().addKind(mykind).build();
- List<Query> mockSplits = new ArrayList<>();
- for (int i = 0; i < 2; i++) {
- mockSplits.add(
- Query.newBuilder()
- .addKind(mykind)
- .setFilter(
- DatastoreHelper.makeFilter("foo", PropertyFilter.Operator.EQUAL,
- Value.newBuilder().setIntegerValue(i).build()))
- .build());
- }
-
- QuerySplitter splitter = mock(QuerySplitter.class);
- when(splitter.getSplits(any(Query.class), any(PartitionId.class), eq(2), any(Datastore.class)))
- .thenReturn(mockSplits);
-
- DatastoreIO.Source io = initialSource
- .withQuery(query)
- .withMockSplitter(splitter)
- .withMockEstimateSizeBytes(8 * 1024L);
-
- DatastoreIO.Source spiedIo = spy(io);
- when(spiedIo.getEstimatedSizeBytes(any(PipelineOptions.class))).thenThrow(new IOException());
-
- List<DatastoreIO.Source> bundles = spiedIo.splitIntoBundles(1024, testPipelineOptions(2));
- assertEquals(2, bundles.size());
- for (int i = 0; i < 2; ++i) {
- DatastoreIO.Source bundle = bundles.get(i);
- Query bundleQuery = bundle.getQuery();
- assertEquals("mykind", bundleQuery.getKind(0).getName());
- assertEquals(i, bundleQuery.getFilter().getPropertyFilter().getValue().getIntegerValue());
- }
- }
-
- @Test
- public void testQuerySplitNoWorkers() throws Exception {
- KindExpression mykind = KindExpression.newBuilder().setName("mykind").build();
- Query query = Query.newBuilder().addKind(mykind).build();
-
List<Query> mockSplits = Lists.newArrayList(Query.newBuilder().addKind(mykind).build());
QuerySplitter splitter = mock(QuerySplitter.class);
@@ -424,7 +380,7 @@ public class DatastoreIOTest {
when(spiedIo.getEstimatedSizeBytes(any(PipelineOptions.class)))
.thenThrow(new NoSuchElementException());
- List<DatastoreIO.Source> bundles = spiedIo.splitIntoBundles(1024, testPipelineOptions(0));
+ List<DatastoreIO.Source> bundles = spiedIo.splitIntoBundles(1024, testPipelineOptions());
assertEquals(1, bundles.size());
verify(splitter, never())
.getSplits(any(Query.class), any(PartitionId.class), eq(1), any(Datastore.class));
[03/12] incubator-beam git commit: [BEAM-151] Break out Dataflow
runner specific tests to separate file
Posted by lc...@apache.org.
[BEAM-151] Break out Dataflow runner specific tests to separate file
This prevents moving the Dataflow runner code to its own separate maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/22fd1206
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/22fd1206
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/22fd1206
Branch: refs/heads/master
Commit: 22fd12061ebb6de921cd6fc22d6eb6e937f9a47d
Parents: 05f7aaa
Author: Luke Cwik <lc...@google.com>
Authored: Fri Mar 25 16:28:36 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:44 2016 -0700
----------------------------------------------------------------------
.../dataflow/sdk/io/DataflowTextIOTest.java | 117 +++++++++++++++++++
.../cloud/dataflow/sdk/io/TextIOTest.java | 79 +------------
2 files changed, 120 insertions(+), 76 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/22fd1206/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
new file mode 100644
index 0000000..8c0d6d2
--- /dev/null
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/DataflowTextIOTest.java
@@ -0,0 +1,117 @@
+/*
+ * Copyright (C) 2015 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.io;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.testing.TestDataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.util.GcsUtil;
+import com.google.cloud.dataflow.sdk.util.TestCredential;
+import com.google.cloud.dataflow.sdk.util.gcsfs.GcsPath;
+import com.google.common.collect.ImmutableList;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+import org.mockito.Mockito;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
+
+import java.io.IOException;
+import java.nio.channels.FileChannel;
+import java.nio.channels.SeekableByteChannel;
+import java.nio.file.Files;
+import java.nio.file.StandardOpenOption;
+import java.util.List;
+
+/**
+ * {@link DataflowPipelineRunner} specific tests for TextIO Read and Write transforms.
+ */
+@RunWith(JUnit4.class)
+public class DataflowTextIOTest {
+
+ private TestDataflowPipelineOptions buildTestPipelineOptions() {
+ TestDataflowPipelineOptions options =
+ PipelineOptionsFactory.as(TestDataflowPipelineOptions.class);
+ options.setGcpCredential(new TestCredential());
+ return options;
+ }
+
+ private GcsUtil buildMockGcsUtil() throws IOException {
+ GcsUtil mockGcsUtil = Mockito.mock(GcsUtil.class);
+
+ // Any request to open gets a new bogus channel
+ Mockito
+ .when(mockGcsUtil.open(Mockito.any(GcsPath.class)))
+ .then(new Answer<SeekableByteChannel>() {
+ @Override
+ public SeekableByteChannel answer(InvocationOnMock invocation) throws Throwable {
+ return FileChannel.open(
+ Files.createTempFile("channel-", ".tmp"),
+ StandardOpenOption.CREATE, StandardOpenOption.DELETE_ON_CLOSE);
+ }
+ });
+
+ // Any request for expansion returns a list containing the original GcsPath
+ // This is required to pass validation that occurs in TextIO during apply()
+ Mockito
+ .when(mockGcsUtil.expand(Mockito.any(GcsPath.class)))
+ .then(new Answer<List<GcsPath>>() {
+ @Override
+ public List<GcsPath> answer(InvocationOnMock invocation) throws Throwable {
+ return ImmutableList.of((GcsPath) invocation.getArguments()[0]);
+ }
+ });
+
+ return mockGcsUtil;
+ }
+
+ /**
+ * This tests a few corner cases that should not crash.
+ */
+ @Test
+ public void testGoodWildcards() throws Exception {
+ TestDataflowPipelineOptions options = buildTestPipelineOptions();
+ options.setGcsUtil(buildMockGcsUtil());
+
+ Pipeline pipeline = Pipeline.create(options);
+
+ applyRead(pipeline, "gs://bucket/foo");
+ applyRead(pipeline, "gs://bucket/foo/");
+ applyRead(pipeline, "gs://bucket/foo/*");
+ applyRead(pipeline, "gs://bucket/foo/?");
+ applyRead(pipeline, "gs://bucket/foo/[0-9]");
+ applyRead(pipeline, "gs://bucket/foo/*baz*");
+ applyRead(pipeline, "gs://bucket/foo/*baz?");
+ applyRead(pipeline, "gs://bucket/foo/[0-9]baz?");
+ applyRead(pipeline, "gs://bucket/foo/baz/*");
+ applyRead(pipeline, "gs://bucket/foo/baz/*wonka*");
+ applyRead(pipeline, "gs://bucket/foo/*baz/wonka*");
+ applyRead(pipeline, "gs://bucket/foo*/baz");
+ applyRead(pipeline, "gs://bucket/foo?/baz");
+ applyRead(pipeline, "gs://bucket/foo[0-9]/baz");
+
+ // Check that running doesn't fail.
+ pipeline.run();
+ }
+
+ private void applyRead(Pipeline pipeline, String path) {
+ pipeline.apply("Read(" + path + ")", TextIO.Read.from(path));
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/22fd1206/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
index 6d4d6e7..a17d153 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/io/TextIOTest.java
@@ -37,15 +37,17 @@ import com.google.cloud.dataflow.sdk.io.TextIO.TextSource;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
import com.google.cloud.dataflow.sdk.testing.SourceTestUtils;
-import com.google.cloud.dataflow.sdk.testing.TestDataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
import com.google.cloud.dataflow.sdk.transforms.Create;
import com.google.cloud.dataflow.sdk.transforms.PTransform;
import com.google.cloud.dataflow.sdk.util.CoderUtils;
+<<<<<<< HEAD
import com.google.cloud.dataflow.sdk.util.GcsUtil;
import com.google.cloud.dataflow.sdk.util.IOChannelUtils;
import com.google.cloud.dataflow.sdk.util.TestCredential;
import com.google.cloud.dataflow.sdk.util.gcsfs.GcsPath;
+=======
+>>>>>>> a8a3419... [BEAM-151] Break out Dataflow runner specific tests to separate file
import com.google.cloud.dataflow.sdk.values.PCollection;
import com.google.cloud.dataflow.sdk.values.PDone;
import com.google.common.collect.ImmutableList;
@@ -56,9 +58,6 @@ import org.junit.rules.ExpectedException;
import org.junit.rules.TemporaryFolder;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-import org.mockito.Mockito;
-import org.mockito.invocation.InvocationOnMock;
-import org.mockito.stubbing.Answer;
import java.io.BufferedReader;
import java.io.File;
@@ -66,11 +65,8 @@ import java.io.FileOutputStream;
import java.io.FileReader;
import java.io.IOException;
import java.io.PrintStream;
-import java.nio.channels.FileChannel;
-import java.nio.channels.SeekableByteChannel;
import java.nio.charset.StandardCharsets;
import java.nio.file.Files;
-import java.nio.file.StandardOpenOption;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.List;
@@ -85,42 +81,6 @@ public class TextIOTest {
@Rule public TemporaryFolder tmpFolder = new TemporaryFolder();
@Rule public ExpectedException expectedException = ExpectedException.none();
- private GcsUtil buildMockGcsUtil() throws IOException {
- GcsUtil mockGcsUtil = Mockito.mock(GcsUtil.class);
-
- // Any request to open gets a new bogus channel
- Mockito
- .when(mockGcsUtil.open(Mockito.any(GcsPath.class)))
- .then(new Answer<SeekableByteChannel>() {
- @Override
- public SeekableByteChannel answer(InvocationOnMock invocation) throws Throwable {
- return FileChannel.open(
- Files.createTempFile("channel-", ".tmp"),
- StandardOpenOption.CREATE, StandardOpenOption.DELETE_ON_CLOSE);
- }
- });
-
- // Any request for expansion returns a list containing the original GcsPath
- // This is required to pass validation that occurs in TextIO during apply()
- Mockito
- .when(mockGcsUtil.expand(Mockito.any(GcsPath.class)))
- .then(new Answer<List<GcsPath>>() {
- @Override
- public List<GcsPath> answer(InvocationOnMock invocation) throws Throwable {
- return ImmutableList.of((GcsPath) invocation.getArguments()[0]);
- }
- });
-
- return mockGcsUtil;
- }
-
- private TestDataflowPipelineOptions buildTestPipelineOptions() {
- TestDataflowPipelineOptions options =
- PipelineOptionsFactory.as(TestDataflowPipelineOptions.class);
- options.setGcpCredential(new TestCredential());
- return options;
- }
-
<T> void runTestRead(T[] expected, Coder<T> coder) throws Exception {
File tmpFile = tmpFolder.newFile("file.txt");
String filename = tmpFile.getPath();
@@ -329,39 +289,6 @@ public class TextIOTest {
}
/**
- * This tests a few corner cases that should not crash.
- */
- @Test
- public void testGoodWildcards() throws Exception {
- TestDataflowPipelineOptions options = buildTestPipelineOptions();
- options.setGcsUtil(buildMockGcsUtil());
-
- Pipeline pipeline = Pipeline.create(options);
-
- applyRead(pipeline, "gs://bucket/foo");
- applyRead(pipeline, "gs://bucket/foo/");
- applyRead(pipeline, "gs://bucket/foo/*");
- applyRead(pipeline, "gs://bucket/foo/?");
- applyRead(pipeline, "gs://bucket/foo/[0-9]");
- applyRead(pipeline, "gs://bucket/foo/*baz*");
- applyRead(pipeline, "gs://bucket/foo/*baz?");
- applyRead(pipeline, "gs://bucket/foo/[0-9]baz?");
- applyRead(pipeline, "gs://bucket/foo/baz/*");
- applyRead(pipeline, "gs://bucket/foo/baz/*wonka*");
- applyRead(pipeline, "gs://bucket/foo/*baz/wonka*");
- applyRead(pipeline, "gs://bucket/foo*/baz");
- applyRead(pipeline, "gs://bucket/foo?/baz");
- applyRead(pipeline, "gs://bucket/foo[0-9]/baz");
-
- // Check that running doesn't fail.
- pipeline.run();
- }
-
- private void applyRead(Pipeline pipeline, String path) {
- pipeline.apply("Read(" + path + ")", TextIO.Read.from(path));
- }
-
- /**
* Recursive wildcards are not supported.
* This tests "**".
*/
[04/12] incubator-beam git commit: [BEAM-151] Break out Dataflow
transport creation to another file
Posted by lc...@apache.org.
[BEAM-151] Break out Dataflow transport creation to another file
This prevents moving DataflowPipelineOptions into a Dataflow runner maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/d0db477a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/d0db477a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/d0db477a
Branch: refs/heads/master
Commit: d0db477a0ce436728f71f0f4aec0b0098eac66be
Parents: c8cb55a
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 13:09:33 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../options/DataflowPipelineDebugOptions.java | 5 +-
.../sdk/runners/DataflowPipelineRunner.java | 6 +-
.../dataflow/sdk/util/DataflowTransport.java | 111 +++++++++++++++++++
.../cloud/dataflow/sdk/util/Transport.java | 46 --------
.../sdk/options/GoogleApiDebugOptionsTest.java | 93 ++++++++--------
5 files changed, 163 insertions(+), 98 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d0db477a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
index b55fa17..6231bd4 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
@@ -20,11 +20,11 @@ package com.google.cloud.dataflow.sdk.options;
import com.google.api.services.dataflow.Dataflow;
import com.google.cloud.dataflow.sdk.annotations.Experimental;
import com.google.cloud.dataflow.sdk.util.DataflowPathValidator;
+import com.google.cloud.dataflow.sdk.util.DataflowTransport;
import com.google.cloud.dataflow.sdk.util.GcsStager;
import com.google.cloud.dataflow.sdk.util.InstanceBuilder;
import com.google.cloud.dataflow.sdk.util.PathValidator;
import com.google.cloud.dataflow.sdk.util.Stager;
-import com.google.cloud.dataflow.sdk.util.Transport;
import com.fasterxml.jackson.annotation.JsonIgnore;
@@ -159,7 +159,8 @@ public interface DataflowPipelineDebugOptions extends PipelineOptions {
public static class DataflowClientFactory implements DefaultValueFactory<Dataflow> {
@Override
public Dataflow create(PipelineOptions options) {
- return Transport.newDataflowClient(options.as(DataflowPipelineOptions.class)).build();
+ return DataflowTransport.newDataflowClient(
+ options.as(DataflowPipelineOptions.class)).build();
}
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d0db477a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java
index 5f43cc3..50ca36f 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java
@@ -99,6 +99,7 @@ import com.google.cloud.dataflow.sdk.transforms.windowing.GlobalWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
import com.google.cloud.dataflow.sdk.util.CoderUtils;
import com.google.cloud.dataflow.sdk.util.DataflowReleaseInfo;
+import com.google.cloud.dataflow.sdk.util.DataflowTransport;
import com.google.cloud.dataflow.sdk.util.IOChannelUtils;
import com.google.cloud.dataflow.sdk.util.InstanceBuilder;
import com.google.cloud.dataflow.sdk.util.MonitoringUtil;
@@ -107,7 +108,6 @@ import com.google.cloud.dataflow.sdk.util.PathValidator;
import com.google.cloud.dataflow.sdk.util.PropertyNames;
import com.google.cloud.dataflow.sdk.util.Reshuffle;
import com.google.cloud.dataflow.sdk.util.SystemDoFnInternal;
-import com.google.cloud.dataflow.sdk.util.Transport;
import com.google.cloud.dataflow.sdk.util.ValueWithRecordId;
import com.google.cloud.dataflow.sdk.util.WindowedValue;
import com.google.cloud.dataflow.sdk.util.WindowedValue.FullWindowedValueCoder;
@@ -444,7 +444,7 @@ public class DataflowPipelineRunner extends PipelineRunner<DataflowPipelineJob>
throw new RuntimeException("Should not specify the debuggee");
}
- Clouddebugger debuggerClient = Transport.newClouddebuggerClient(options).build();
+ Clouddebugger debuggerClient = DataflowTransport.newClouddebuggerClient(options).build();
Debuggee debuggee = registerDebuggee(debuggerClient, uniquifier);
options.setDebuggee(debuggee);
@@ -600,7 +600,7 @@ public class DataflowPipelineRunner extends PipelineRunner<DataflowPipelineJob>
// regularly and need not be retried automatically.
DataflowPipelineJob dataflowPipelineJob =
new DataflowPipelineJob(options.getProject(), jobResult.getId(),
- Transport.newRawDataflowClient(options).build(), aggregatorTransforms);
+ DataflowTransport.newRawDataflowClient(options).build(), aggregatorTransforms);
// If the service returned client request id, the SDK needs to compare it
// with the original id generated in the request, if they are not the same
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d0db477a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
new file mode 100644
index 0000000..8de358c
--- /dev/null
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/DataflowTransport.java
@@ -0,0 +1,111 @@
+/*******************************************************************************
+ * Copyright (C) 2015 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.util;
+
+import static com.google.cloud.dataflow.sdk.util.Transport.getJsonFactory;
+import static com.google.cloud.dataflow.sdk.util.Transport.getTransport;
+
+import com.google.api.client.auth.oauth2.Credential;
+import com.google.api.client.http.HttpRequestInitializer;
+import com.google.api.services.clouddebugger.v2.Clouddebugger;
+import com.google.api.services.dataflow.Dataflow;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer;
+import com.google.common.collect.ImmutableList;
+
+import java.net.MalformedURLException;
+import java.net.URL;
+
+/**
+ * Helpers for cloud communication.
+ */
+public class DataflowTransport {
+
+
+ private static class ApiComponents {
+ public String rootUrl;
+ public String servicePath;
+
+ public ApiComponents(String root, String path) {
+ this.rootUrl = root;
+ this.servicePath = path;
+ }
+ }
+
+ private static ApiComponents apiComponentsFromUrl(String urlString) {
+ try {
+ URL url = new URL(urlString);
+ String rootUrl = url.getProtocol() + "://" + url.getHost() +
+ (url.getPort() > 0 ? ":" + url.getPort() : "");
+ return new ApiComponents(rootUrl, url.getPath());
+ } catch (MalformedURLException e) {
+ throw new RuntimeException("Invalid URL: " + urlString);
+ }
+ }
+
+ /**
+ * Returns a Google Cloud Dataflow client builder.
+ */
+ public static Dataflow.Builder newDataflowClient(DataflowPipelineOptions options) {
+ String servicePath = options.getDataflowEndpoint();
+ ApiComponents components;
+ if (servicePath.contains("://")) {
+ components = apiComponentsFromUrl(servicePath);
+ } else {
+ components = new ApiComponents(options.getApiRootUrl(), servicePath);
+ }
+
+ return new Dataflow.Builder(getTransport(),
+ getJsonFactory(),
+ chainHttpRequestInitializer(
+ options.getGcpCredential(),
+ // Do not log 404. It clutters the output and is possibly even required by the caller.
+ new RetryHttpRequestInitializer(ImmutableList.of(404))))
+ .setApplicationName(options.getAppName())
+ .setRootUrl(components.rootUrl)
+ .setServicePath(components.servicePath)
+ .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
+ }
+
+ public static Clouddebugger.Builder newClouddebuggerClient(DataflowPipelineOptions options) {
+ return new Clouddebugger.Builder(getTransport(),
+ getJsonFactory(),
+ chainHttpRequestInitializer(options.getGcpCredential(), new RetryHttpRequestInitializer()))
+ .setApplicationName(options.getAppName())
+ .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
+ }
+
+ /**
+ * Returns a Dataflow client that does not automatically retry failed
+ * requests.
+ */
+ public static Dataflow.Builder
+ newRawDataflowClient(DataflowPipelineOptions options) {
+ return newDataflowClient(options)
+ .setHttpRequestInitializer(options.getGcpCredential())
+ .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
+ }
+
+ private static HttpRequestInitializer chainHttpRequestInitializer(
+ Credential credential, HttpRequestInitializer httpRequestInitializer) {
+ if (credential == null) {
+ return httpRequestInitializer;
+ } else {
+ return new ChainingHttpRequestInitializer(credential, httpRequestInitializer);
+ }
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d0db477a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
index 187d164..5888822 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
@@ -24,13 +24,10 @@ import com.google.api.client.http.HttpTransport;
import com.google.api.client.json.JsonFactory;
import com.google.api.client.json.jackson2.JacksonFactory;
import com.google.api.services.bigquery.Bigquery;
-import com.google.api.services.clouddebugger.v2.Clouddebugger;
-import com.google.api.services.dataflow.Dataflow;
import com.google.api.services.pubsub.Pubsub;
import com.google.api.services.storage.Storage;
import com.google.cloud.dataflow.sdk.options.BigQueryOptions;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineDebugOptions;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.GcsOptions;
import com.google.cloud.dataflow.sdk.options.PubsubOptions;
import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer;
@@ -128,49 +125,6 @@ public class Transport {
}
/**
- * Returns a Google Cloud Dataflow client builder.
- */
- public static Dataflow.Builder newDataflowClient(DataflowPipelineOptions options) {
- String servicePath = options.getDataflowEndpoint();
- ApiComponents components;
- if (servicePath.contains("://")) {
- components = apiComponentsFromUrl(servicePath);
- } else {
- components = new ApiComponents(options.getApiRootUrl(), servicePath);
- }
-
- return new Dataflow.Builder(getTransport(),
- getJsonFactory(),
- chainHttpRequestInitializer(
- options.getGcpCredential(),
- // Do not log 404. It clutters the output and is possibly even required by the caller.
- new RetryHttpRequestInitializer(ImmutableList.of(404))))
- .setApplicationName(options.getAppName())
- .setRootUrl(components.rootUrl)
- .setServicePath(components.servicePath)
- .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
- }
-
- public static Clouddebugger.Builder newClouddebuggerClient(DataflowPipelineOptions options) {
- return new Clouddebugger.Builder(getTransport(),
- getJsonFactory(),
- chainHttpRequestInitializer(options.getGcpCredential(), new RetryHttpRequestInitializer()))
- .setApplicationName(options.getAppName())
- .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
- }
-
- /**
- * Returns a Dataflow client that does not automatically retry failed
- * requests.
- */
- public static Dataflow.Builder
- newRawDataflowClient(DataflowPipelineOptions options) {
- return newDataflowClient(options)
- .setHttpRequestInitializer(options.getGcpCredential())
- .setGoogleClientRequestInitializer(options.getGoogleApiTrace());
- }
-
- /**
* Returns a Cloud Storage client builder.
*
* <p>Note: this client's endpoint is <b>not</b> modified by the
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/d0db477a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/GoogleApiDebugOptionsTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/GoogleApiDebugOptionsTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/GoogleApiDebugOptionsTest.java
index 3201608..c2f0bb8 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/GoogleApiDebugOptionsTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/GoogleApiDebugOptionsTest.java
@@ -22,8 +22,7 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import com.google.api.services.bigquery.Bigquery.Datasets.Delete;
-import com.google.api.services.dataflow.Dataflow.Projects.Jobs.Create;
-import com.google.api.services.dataflow.Dataflow.Projects.Jobs.Get;
+import com.google.api.services.storage.Storage;
import com.google.cloud.dataflow.sdk.options.GoogleApiDebugOptions.GoogleApiTracer;
import com.google.cloud.dataflow.sdk.util.TestCredential;
import com.google.cloud.dataflow.sdk.util.Transport;
@@ -37,104 +36,104 @@ import org.junit.runners.JUnit4;
/** Tests for {@link GoogleApiDebugOptions}. */
@RunWith(JUnit4.class)
public class GoogleApiDebugOptionsTest {
+ private static final String STORAGE_GET_TRACE =
+ "--googleApiTrace={\"Objects.Get\":\"GetTraceDestination\"}";
+ private static final String STORAGE_GET_AND_LIST_TRACE =
+ "--googleApiTrace={\"Objects.Get\":\"GetTraceDestination\","
+ + "\"Objects.List\":\"ListTraceDestination\"}";
+ private static final String STORAGE_TRACE = "--googleApiTrace={\"Storage\":\"TraceDestination\"}";
+
@Test
public void testWhenTracingMatches() throws Exception {
- String[] args =
- new String[] {"--googleApiTrace={\"Projects.Jobs.Get\":\"GetTraceDestination\"}"};
- DataflowPipelineOptions options =
- PipelineOptionsFactory.fromArgs(args).as(DataflowPipelineOptions.class);
+ String[] args = new String[] {STORAGE_GET_TRACE};
+ GcsOptions options = PipelineOptionsFactory.fromArgs(args).as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
-
assertNotNull(options.getGoogleApiTrace());
- Get request =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.Get request =
+ Transport.newStorageClient(options).build().objects().get("testBucketId", "testObjectId");
assertEquals("GetTraceDestination", request.get("$trace"));
}
@Test
public void testWhenTracingDoesNotMatch() throws Exception {
- String[] args = new String[] {"--googleApiTrace={\"Projects.Jobs.Create\":\"testToken\"}"};
- DataflowPipelineOptions options =
- PipelineOptionsFactory.fromArgs(args).as(DataflowPipelineOptions.class);
+ String[] args = new String[] {STORAGE_GET_TRACE};
+ GcsOptions options = PipelineOptionsFactory.fromArgs(args).as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
assertNotNull(options.getGoogleApiTrace());
- Get request =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.List request =
+ Transport.newStorageClient(options).build().objects().list("testProjectId");
assertNull(request.get("$trace"));
}
@Test
public void testWithMultipleTraces() throws Exception {
- String[] args = new String[] {
- "--googleApiTrace={\"Projects.Jobs.Create\":\"CreateTraceDestination\","
- + "\"Projects.Jobs.Get\":\"GetTraceDestination\"}"};
- DataflowPipelineOptions options =
- PipelineOptionsFactory.fromArgs(args).as(DataflowPipelineOptions.class);
+ String[] args = new String[] {STORAGE_GET_AND_LIST_TRACE};
+ GcsOptions options = PipelineOptionsFactory.fromArgs(args).as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
assertNotNull(options.getGoogleApiTrace());
- Get getRequest =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.Get getRequest =
+ Transport.newStorageClient(options).build().objects().get("testBucketId", "testObjectId");
assertEquals("GetTraceDestination", getRequest.get("$trace"));
- Create createRequest =
- options.getDataflowClient().projects().jobs().create("testProjectId", null);
- assertEquals("CreateTraceDestination", createRequest.get("$trace"));
+ Storage.Objects.List listRequest =
+ Transport.newStorageClient(options).build().objects().list("testProjectId");
+ assertEquals("ListTraceDestination", listRequest.get("$trace"));
}
@Test
- public void testMatchingAllDataflowCalls() throws Exception {
- String[] args = new String[] {"--googleApiTrace={\"Dataflow\":\"TraceDestination\"}"};
- DataflowPipelineOptions options =
- PipelineOptionsFactory.fromArgs(args).as(DataflowPipelineOptions.class);
+ public void testMatchingAllCalls() throws Exception {
+ String[] args = new String[] {STORAGE_TRACE};
+ GcsOptions options =
+ PipelineOptionsFactory.fromArgs(args).as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
assertNotNull(options.getGoogleApiTrace());
- Get getRequest =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.Get getRequest =
+ Transport.newStorageClient(options).build().objects().get("testBucketId", "testObjectId");
assertEquals("TraceDestination", getRequest.get("$trace"));
- Create createRequest =
- options.getDataflowClient().projects().jobs().create("testProjectId", null);
- assertEquals("TraceDestination", createRequest.get("$trace"));
+ Storage.Objects.List listRequest =
+ Transport.newStorageClient(options).build().objects().list("testProjectId");
+ assertEquals("TraceDestination", listRequest.get("$trace"));
}
@Test
public void testMatchingAgainstClient() throws Exception {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ GcsOptions options = PipelineOptionsFactory.as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
options.setGoogleApiTrace(new GoogleApiTracer().addTraceFor(
- Transport.newDataflowClient(options).build(), "TraceDestination"));
+ Transport.newStorageClient(options).build(), "TraceDestination"));
- Get getRequest =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.Get getRequest =
+ Transport.newStorageClient(options).build().objects().get("testBucketId", "testObjectId");
assertEquals("TraceDestination", getRequest.get("$trace"));
- Delete deleteRequest = Transport.newBigQueryClient(options).build().datasets()
- .delete("testProjectId", "testDatasetId");
+ Delete deleteRequest = Transport.newBigQueryClient(options.as(BigQueryOptions.class))
+ .build().datasets().delete("testProjectId", "testDatasetId");
assertNull(deleteRequest.get("$trace"));
}
@Test
public void testMatchingAgainstRequestType() throws Exception {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ GcsOptions options = PipelineOptionsFactory.as(GcsOptions.class);
options.setGcpCredential(new TestCredential());
options.setGoogleApiTrace(new GoogleApiTracer().addTraceFor(
- Transport.newDataflowClient(options).build().projects().jobs()
- .get("aProjectId", "aJobId"), "TraceDestination"));
+ Transport.newStorageClient(options).build().objects()
+ .get("aProjectId", "aObjectId"), "TraceDestination"));
- Get getRequest =
- options.getDataflowClient().projects().jobs().get("testProjectId", "testJobId");
+ Storage.Objects.Get getRequest =
+ Transport.newStorageClient(options).build().objects().get("testBucketId", "testObjectId");
assertEquals("TraceDestination", getRequest.get("$trace"));
- Create createRequest =
- options.getDataflowClient().projects().jobs().create("testProjectId", null);
- assertNull(createRequest.get("$trace"));
+ Storage.Objects.List listRequest =
+ Transport.newStorageClient(options).build().objects().list("testProjectId");
+ assertNull(listRequest.get("$trace"));
}
@Test
[02/12] incubator-beam git commit: [BEAM-151] Break out Dataflow
runner dependency to separate test file
Posted by lc...@apache.org.
[BEAM-151] Break out Dataflow runner dependency to separate test file
This allows for moving the Dataflow specific portion of the test to the Dataflow runner maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/a18ab871
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/a18ab871
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/a18ab871
Branch: refs/heads/master
Commit: a18ab871f8f1b9a3219cee2ecd5d519af23efa87
Parents: 7f0a2f7
Author: Luke Cwik <lc...@google.com>
Authored: Fri Mar 25 16:09:18 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:18:32 2016 -0700
----------------------------------------------------------------------
.../sdk/transforms/DataflowGroupByKeyTest.java | 110 +++++++++++++++++++
.../dataflow/sdk/transforms/GroupByKeyTest.java | 62 -----------
2 files changed, 110 insertions(+), 62 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a18ab871/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
new file mode 100644
index 0000000..b05e7a2
--- /dev/null
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/DataflowGroupByKeyTest.java
@@ -0,0 +1,110 @@
+/*
+ * Copyright (C) 2016 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.transforms;
+
+import com.google.cloud.dataflow.sdk.Pipeline;
+import com.google.cloud.dataflow.sdk.coders.BigEndianIntegerCoder;
+import com.google.cloud.dataflow.sdk.coders.KvCoder;
+import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
+import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
+import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Sessions;
+import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
+import com.google.cloud.dataflow.sdk.util.NoopPathValidator;
+import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
+import com.google.cloud.dataflow.sdk.values.KV;
+import com.google.cloud.dataflow.sdk.values.PBegin;
+import com.google.cloud.dataflow.sdk.values.PCollection;
+import com.google.cloud.dataflow.sdk.values.TypeDescriptor;
+
+import org.joda.time.Duration;
+import org.junit.Rule;
+import org.junit.Test;
+import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.junit.runners.JUnit4;
+
+import java.util.Arrays;
+import java.util.List;
+
+/** Tests for {@link GroupByKey} for the {@link DataflowPipelineRunner}. */
+@RunWith(JUnit4.class)
+public class DataflowGroupByKeyTest {
+ @Rule
+ public ExpectedException thrown = ExpectedException.none();
+
+ /**
+ * Create a test pipeline that uses the {@link DataflowPipelineRunner} so that {@link GroupByKey}
+ * is not expanded. This is used for verifying that even without expansion the proper errors show
+ * up.
+ */
+ private Pipeline createTestServiceRunner() {
+ DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
+ options.setRunner(DataflowPipelineRunner.class);
+ options.setProject("someproject");
+ options.setStagingLocation("gs://staging");
+ options.setPathValidatorClass(NoopPathValidator.class);
+ options.setDataflowClient(null);
+ return Pipeline.create(options);
+ }
+
+ @Test
+ public void testInvalidWindowsService() {
+ Pipeline p = createTestServiceRunner();
+
+ List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
+
+ PCollection<KV<String, Integer>> input =
+ p.apply(Create.of(ungroupedPairs)
+ .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of())))
+ .apply(Window.<KV<String, Integer>>into(
+ Sessions.withGapDuration(Duration.standardMinutes(1))));
+
+ thrown.expect(IllegalStateException.class);
+ thrown.expectMessage("GroupByKey must have a valid Window merge function");
+ input
+ .apply("GroupByKey", GroupByKey.<String, Integer>create())
+ .apply("GroupByKeyAgain", GroupByKey.<String, Iterable<Integer>>create());
+ }
+
+ @Test
+ public void testGroupByKeyServiceUnbounded() {
+ Pipeline p = createTestServiceRunner();
+
+ PCollection<KV<String, Integer>> input =
+ p.apply(
+ new PTransform<PBegin, PCollection<KV<String, Integer>>>() {
+ @Override
+ public PCollection<KV<String, Integer>> apply(PBegin input) {
+ return PCollection.<KV<String, Integer>>createPrimitiveOutputInternal(
+ input.getPipeline(),
+ WindowingStrategy.globalDefault(),
+ PCollection.IsBounded.UNBOUNDED)
+ .setTypeDescriptorInternal(new TypeDescriptor<KV<String, Integer>>() {});
+ }
+ });
+
+ thrown.expect(IllegalStateException.class);
+ thrown.expectMessage(
+ "GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without "
+ + "a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.");
+
+ input.apply("GroupByKey", GroupByKey.<String, Integer>create());
+ }
+}
+
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/a18ab871/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java
index bb64f60..6fb811e 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java
@@ -28,10 +28,8 @@ import com.google.cloud.dataflow.sdk.coders.BigEndianIntegerCoder;
import com.google.cloud.dataflow.sdk.coders.KvCoder;
import com.google.cloud.dataflow.sdk.coders.MapCoder;
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.DirectPipelineOptions;
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
@@ -41,7 +39,6 @@ import com.google.cloud.dataflow.sdk.transforms.windowing.InvalidWindows;
import com.google.cloud.dataflow.sdk.transforms.windowing.OutputTimeFns;
import com.google.cloud.dataflow.sdk.transforms.windowing.Sessions;
import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
-import com.google.cloud.dataflow.sdk.util.NoopPathValidator;
import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
import com.google.cloud.dataflow.sdk.values.KV;
import com.google.cloud.dataflow.sdk.values.PBegin;
@@ -241,21 +238,6 @@ public class GroupByKeyTest {
Duration.standardMinutes(1)))));
}
- /**
- * Create a test pipeline that uses the {@link DataflowPipelineRunner} so that {@link GroupByKey}
- * is not expanded. This is used for verifying that even without expansion the proper errors show
- * up.
- */
- private Pipeline createTestServiceRunner() {
- DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
- options.setRunner(DataflowPipelineRunner.class);
- options.setProject("someproject");
- options.setStagingLocation("gs://staging");
- options.setPathValidatorClass(NoopPathValidator.class);
- options.setDataflowClient(null);
- return Pipeline.create(options);
- }
-
private Pipeline createTestDirectRunner() {
DirectPipelineOptions options = PipelineOptionsFactory.as(DirectPipelineOptions.class);
options.setRunner(DirectPipelineRunner.class);
@@ -282,25 +264,6 @@ public class GroupByKeyTest {
}
@Test
- public void testInvalidWindowsService() {
- Pipeline p = createTestServiceRunner();
-
- List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
-
- PCollection<KV<String, Integer>> input =
- p.apply(Create.of(ungroupedPairs)
- .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of())))
- .apply(Window.<KV<String, Integer>>into(
- Sessions.withGapDuration(Duration.standardMinutes(1))));
-
- thrown.expect(IllegalStateException.class);
- thrown.expectMessage("GroupByKey must have a valid Window merge function");
- input
- .apply("GroupByKey", GroupByKey.<String, Integer>create())
- .apply("GroupByKeyAgain", GroupByKey.<String, Iterable<Integer>>create());
- }
-
- @Test
public void testRemerge() {
Pipeline p = TestPipeline.create();
@@ -350,31 +313,6 @@ public class GroupByKeyTest {
input.apply("GroupByKey", GroupByKey.<String, Integer>create());
}
- @Test
- public void testGroupByKeyServiceUnbounded() {
- Pipeline p = createTestServiceRunner();
-
- PCollection<KV<String, Integer>> input =
- p.apply(
- new PTransform<PBegin, PCollection<KV<String, Integer>>>() {
- @Override
- public PCollection<KV<String, Integer>> apply(PBegin input) {
- return PCollection.<KV<String, Integer>>createPrimitiveOutputInternal(
- input.getPipeline(),
- WindowingStrategy.globalDefault(),
- PCollection.IsBounded.UNBOUNDED)
- .setTypeDescriptorInternal(new TypeDescriptor<KV<String, Integer>>() {});
- }
- });
-
- thrown.expect(IllegalStateException.class);
- thrown.expectMessage(
- "GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without "
- + "a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.");
-
- input.apply("GroupByKey", GroupByKey.<String, Integer>create());
- }
-
/**
* Tests that when two elements are combined via a GroupByKey their output timestamp agrees
* with the windowing function customized to actually be the same as the default, the earlier of
[07/12] incubator-beam git commit: [BEAM-151] Remove references to
Dataflow runners
Posted by lc...@apache.org.
[BEAM-151] Remove references to Dataflow runners
This prevents moving DataflowPipelineRunner and subclasses to runner specific maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/49d65946
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/49d65946
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/49d65946
Branch: refs/heads/master
Commit: 49d65946e4b11afe50a926e9846c4e12b1529aa3
Parents: c8f09a0
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 13:26:19 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../dataflow/sdk/options/PipelineOptionsFactoryTest.java | 11 +++++------
1 file changed, 5 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/49d65946/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactoryTest.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactoryTest.java b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactoryTest.java
index d9d62c8..d8ba8e3 100644
--- a/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactoryTest.java
+++ b/sdks/java/core/src/test/java/com/google/cloud/dataflow/sdk/options/PipelineOptionsFactoryTest.java
@@ -29,8 +29,6 @@ import static org.junit.Assert.assertTrue;
import com.google.cloud.dataflow.sdk.Pipeline;
import com.google.cloud.dataflow.sdk.PipelineResult;
-import com.google.cloud.dataflow.sdk.runners.BlockingDataflowPipelineRunner;
-import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
import com.google.cloud.dataflow.sdk.testing.ExpectedLogs;
@@ -829,20 +827,21 @@ public class PipelineOptionsFactoryTest {
@Test
public void testSettingRunner() {
- String[] args = new String[] {"--runner=BlockingDataflowPipelineRunner"};
+ String[] args = new String[] {"--runner=DirectPipelineRunner"};
PipelineOptions options = PipelineOptionsFactory.fromArgs(args).create();
- assertEquals(BlockingDataflowPipelineRunner.class, options.getRunner());
+ assertEquals(DirectPipelineRunner.class, options.getRunner());
}
@Test
public void testSettingRunnerFullName() {
String[] args =
- new String[] {String.format("--runner=%s", DataflowPipelineRunner.class.getName())};
+ new String[] {String.format("--runner=%s", DirectPipelineRunner.class.getName())};
PipelineOptions opts = PipelineOptionsFactory.fromArgs(args).create();
- assertEquals(opts.getRunner(), DataflowPipelineRunner.class);
+ assertEquals(opts.getRunner(), DirectPipelineRunner.class);
}
+
@Test
public void testSettingUnknownRunner() {
String[] args = new String[] {"--runner=UnknownRunner"};
[09/12] incubator-beam git commit: [BEAM-151] Migrate options
relevant to pubsub to its own options interface
Posted by lc...@apache.org.
[BEAM-151] Migrate options relevant to pubsub to its own options interface
This prevents moving DataflowPipelineOptions to the Dataflow runner maven module.
Project: http://git-wip-us.apache.org/repos/asf/incubator-beam/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-beam/commit/c8cb55af
Tree: http://git-wip-us.apache.org/repos/asf/incubator-beam/tree/c8cb55af
Diff: http://git-wip-us.apache.org/repos/asf/incubator-beam/diff/c8cb55af
Branch: refs/heads/master
Commit: c8cb55af49832f39788e1936a9576760fd720eb1
Parents: a502baa
Author: Luke Cwik <lc...@google.com>
Authored: Mon Mar 28 10:02:01 2016 -0700
Committer: Luke Cwik <lc...@google.com>
Committed: Thu Apr 7 11:19:49 2016 -0700
----------------------------------------------------------------------
.../google/cloud/dataflow/sdk/io/PubsubIO.java | 6 ++--
.../options/DataflowPipelineDebugOptions.java | 8 -----
.../sdk/options/DataflowPipelineOptions.java | 2 +-
.../dataflow/sdk/options/PubsubOptions.java | 35 ++++++++++++++++++++
.../cloud/dataflow/sdk/util/Transport.java | 3 +-
5 files changed, 41 insertions(+), 13 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
index b7f2afe..68615c5 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/io/PubsubIO.java
@@ -33,7 +33,7 @@ import com.google.api.services.pubsub.model.Subscription;
import com.google.cloud.dataflow.sdk.coders.Coder;
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
import com.google.cloud.dataflow.sdk.coders.VoidCoder;
-import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
+import com.google.cloud.dataflow.sdk.options.PubsubOptions;
import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
import com.google.cloud.dataflow.sdk.runners.PipelineRunner;
@@ -728,7 +728,7 @@ public class PubsubIO {
@Override
public void processElement(ProcessContext c) throws IOException {
Pubsub pubsubClient =
- Transport.newPubsubClient(c.getPipelineOptions().as(DataflowPipelineOptions.class))
+ Transport.newPubsubClient(c.getPipelineOptions().as(PubsubOptions.class))
.build();
String subscription;
@@ -1004,7 +1004,7 @@ public class PubsubIO {
public void startBundle(Context c) {
this.output = new ArrayList<>();
this.pubsubClient =
- Transport.newPubsubClient(c.getPipelineOptions().as(DataflowPipelineOptions.class))
+ Transport.newPubsubClient(c.getPipelineOptions().as(PubsubOptions.class))
.build();
}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
index d4f6385..b55fa17 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineDebugOptions.java
@@ -164,14 +164,6 @@ public interface DataflowPipelineDebugOptions extends PipelineOptions {
}
/**
- * Root URL for use with the Pubsub API.
- */
- @Description("Root URL for use with the Pubsub API")
- @Default.String("https://pubsub.googleapis.com")
- String getPubsubRootUrl();
- void setPubsubRootUrl(String value);
-
- /**
* Whether to update the currently running pipeline with the same name as this one.
*
* @deprecated This property is replaced by {@link DataflowPipelineOptions#getUpdate()}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java
index 23ca0e7..dbfafd1 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/DataflowPipelineOptions.java
@@ -33,7 +33,7 @@ public interface DataflowPipelineOptions extends
PipelineOptions, GcpOptions, ApplicationNameOptions, DataflowPipelineDebugOptions,
DataflowPipelineWorkerPoolOptions, BigQueryOptions,
GcsOptions, StreamingOptions, CloudDebuggerOptions, DataflowWorkerLoggingOptions,
- DataflowProfilingOptions {
+ DataflowProfilingOptions, PubsubOptions {
@Description("Project id. Required when running a Dataflow in the cloud. "
+ "See https://cloud.google.com/storage/docs/projects for further details.")
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
new file mode 100644
index 0000000..51cf62c
--- /dev/null
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/options/PubsubOptions.java
@@ -0,0 +1,35 @@
+/*
+ * Copyright (C) 2016 Google Inc.
+ *
+ * Licensed 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 com.google.cloud.dataflow.sdk.options;
+
+/**
+ * Properties that can be set when using Pubsub with the Dataflow SDK.
+ */
+@Description("Options that are used to configure BigQuery. See "
+ + "https://cloud.google.com/bigquery/what-is-bigquery for details on BigQuery.")
+public interface PubsubOptions extends ApplicationNameOptions, GcpOptions,
+ PipelineOptions, StreamingOptions {
+
+ /**
+ * Root URL for use with the Pubsub API.
+ */
+ @Description("Root URL for use with the Pubsub API")
+ @Default.String("https://pubsub.googleapis.com")
+ @Hidden
+ String getPubsubRootUrl();
+ void setPubsubRootUrl(String value);
+}
http://git-wip-us.apache.org/repos/asf/incubator-beam/blob/c8cb55af/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
----------------------------------------------------------------------
diff --git a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
index 27f61fa..187d164 100644
--- a/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
+++ b/sdks/java/core/src/main/java/com/google/cloud/dataflow/sdk/util/Transport.java
@@ -32,6 +32,7 @@ import com.google.cloud.dataflow.sdk.options.BigQueryOptions;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineDebugOptions;
import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
import com.google.cloud.dataflow.sdk.options.GcsOptions;
+import com.google.cloud.dataflow.sdk.options.PubsubOptions;
import com.google.cloud.hadoop.util.ChainingHttpRequestInitializer;
import com.google.common.collect.ImmutableList;
@@ -115,7 +116,7 @@ public class Transport {
* {@link DataflowPipelineDebugOptions#getApiRootUrl()} option.
*/
public static Pubsub.Builder
- newPubsubClient(DataflowPipelineOptions options) {
+ newPubsubClient(PubsubOptions options) {
return new Pubsub.Builder(getTransport(), getJsonFactory(),
chainHttpRequestInitializer(
options.getGcpCredential(),