FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

Move GroupByKey validation into validate method · imace/DataflowJavaSDK@39ba987 · GitHub

  1. imace
  2. DataflowJavaSDK

Repository navigation

Commit 39ba987

Browse files
bchambers
authored and
davorbonaci
committed
Move GroupByKey validation into validate method
----Release Notes---- Re-enabled verification of GroupByKey usage. Specififically, the key must have a deterministic coder and using GroupByKey with an Unbounded PCollection requires windowing or triggers. [] ------------- Created by MOE: http://code.google.com/p/moe-java MOE_MIGRATED_REVID=99194010
1 parent ccd4e0a commit 39ba987

4 files changed

Lines changed: 161 additions & 62 deletions

File tree

‎sdk/src/main/java/com/google/cloud/dataflow/sdk/runners/DataflowPipelineRunner.java‎

Lines changed: 6 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -73,6 +73,7 @@
7373
import com.google.cloud.dataflow.sdk.values.PDone;
7474
import com.google.cloud.dataflow.sdk.values.PInput;
7575
import com.google.cloud.dataflow.sdk.values.POutput;
76+
import com.google.common.annotations.VisibleForTesting;
7677
import com.google.common.base.Preconditions;
7778
import com.google.common.base.Strings;
7879
import com.google.common.collect.ImmutableMap;
@@ -188,7 +189,7 @@ public static DataflowPipelineRunner fromOptions(PipelineOptions options) {
188189
return new DataflowPipelineRunner(dataflowOptions);
189190
}
190191

191-
private DataflowPipelineRunner(DataflowPipelineOptions options) {
192+
@VisibleForTesting protected DataflowPipelineRunner(DataflowPipelineOptions options) {
192193
this.options = options;
193194
this.dataflowClient = options.getDataflowClient();
194195
this.translator = DataflowPipelineTranslator.fromOptions(options);
@@ -224,7 +225,9 @@ public <OutputT extends POutput, InputT extends PInput> OutputT apply(
224225
@SuppressWarnings("unchecked")
225226
OutputT outputT = (OutputT) PCollection.createPrimitiveOutputInternal(
226227
pc.getPipeline(),
227-
pc.getWindowingStrategy(),
228+
transform instanceof GroupByKey
229+
? ((GroupByKey<?, ?>) transform).updateWindowingStrategy(pc.getWindowingStrategy())
230+
: pc.getWindowingStrategy(),
228231
pc.isBounded());
229232
return outputT;
230233

@@ -662,7 +665,7 @@ public PCollection<T> apply(PInput input) {
662665
.setWindowingStrategyInternal(WindowingStrategy.globalDefault())
663666
.apply(Window.<KV<Void, Iterable<Void>>>into(new GlobalWindows()))
664667
.apply(ParDo.of(new OutputElements<>(transform.getElements(), coder)))
665-
.setCoder(coder);
668+
.setCoder(coder).setIsBoundedInternal(IsBounded.BOUNDED);
666669
} catch (CannotProvideCoderException e) {
667670
throw new IllegalArgumentException("Unable to infer a coder and no Coder was specified. "
668671
+ "Please set a coder by invoking Create.withCoder() explicitly.", e);

‎sdk/src/main/java/com/google/cloud/dataflow/sdk/transforms/GroupByKey.java‎

Lines changed: 52 additions & 41 deletions
Original file line numberDiff line numberDiff line change
@@ -176,18 +176,60 @@ public boolean fewKeys() {
176176
/////////////////////////////////////////////////////////////////////////////
177177

178178
@Override
179-
public PCollection<KV<K, Iterable<V>>> apply(PCollection<KV<K, V>> input) {
180-
// This operation groups by the combination of key and window,
181-
// merging windows as needed, using the windows assigned to the
182-
// key/value input elements and the window merge operation of the
183-
// window function associated with the input PCollection.
179+
public void validate(PCollection<KV<K, V>> input) {
184180
WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
181+
// Verify that the input PCollection is bounded, or that there is windowing/triggering being
182+
// used. Without this, the watermark (at end of global window) will never be reached.
183+
if (windowingStrategy.getWindowFn() instanceof GlobalWindows
184+
&& windowingStrategy.getTrigger().getSpec() instanceof DefaultTrigger
185+
&& input.isBounded() != IsBounded.BOUNDED) {
186+
throw new IllegalStateException("GroupByKey cannot be applied to non-bounded PCollection in "
187+
+ "the GlobalWindow without a trigger. Use a Window.into or Window.triggering transform "
188+
+ "prior to GroupByKey.");
189+
}
190+
191+
// Verify that the input Coder<KV<K, V>> is a KvCoder<K, V>, and that
192+
// the key coder is deterministic.
193+
Coder<K> keyCoder = getKeyCoder(input.getCoder());
194+
try {
195+
keyCoder.verifyDeterministic();
196+
} catch (NonDeterministicException e) {
197+
throw new IllegalStateException(
198+
"the keyCoder of a GroupByKey must be deterministic", e);
199+
}
200+
201+
// Validate the window merge function.
185202
if (windowingStrategy.getWindowFn() instanceof InvalidWindows) {
186203
String cause = ((InvalidWindows<?>) windowingStrategy.getWindowFn()).getCause();
187204
throw new IllegalStateException(
188205
"GroupByKey must have a valid Window merge function. "
189206
+ "Invalid because: " + cause);
190207
}
208+
}
209+
210+
public WindowingStrategy<?, ?> updateWindowingStrategy(WindowingStrategy<?, ?> inputStrategy) {
211+
WindowFn<?, ?> inputWindowFn = inputStrategy.getWindowFn();
212+
if (!inputWindowFn.isNonMerging()) {
213+
// Prevent merging windows again, without explicit user
214+
// involvement, e.g., by Window.into() or Window.remerge().
215+
inputWindowFn = new InvalidWindows<>(
216+
"WindowFn has already been consumed by previous GroupByKey", inputWindowFn);
217+
}
218+
219+
// We also switch to the continuation trigger associated with the current trigger.
220+
return inputStrategy
221+
.withWindowFn(inputWindowFn)
222+
.withTrigger(inputStrategy.getTrigger().getSpec().getContinuationTrigger());
223+
}
224+
225+
@Override
226+
public PCollection<KV<K, Iterable<V>>> apply(PCollection<KV<K, V>> input) {
227+
// This operation groups by the combination of key and window,
228+
// merging windows as needed, using the windows assigned to the
229+
// key/value input elements and the window merge operation of the
230+
// window function associated with the input PCollection.
231+
WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
232+
191233
// By default, implement GroupByKey[AndWindow] via a series of lower-level
192234
// operations.
193235
return input
@@ -206,7 +248,10 @@ public PCollection<KV<K, Iterable<V>>> apply(PCollection<KV<K, V>> input) {
206248
.apply(new SortValuesByTimestamp<K, V>())
207249

208250
// Group each key's values by window, merging windows as needed.
209-
.apply(new GroupAlsoByWindow<K, V>(windowingStrategy));
251+
.apply(new GroupAlsoByWindow<K, V>(windowingStrategy))
252+
253+
// And update the windowing strategy as appropriate.
254+
.setWindowingStrategyInternal(updateWindowingStrategy(windowingStrategy));
210255
}
211256

212257
@Override
@@ -383,46 +428,12 @@ public PCollection<KV<K, Iterable<V>>> apply(
383428
public static class GroupByKeyOnly<K, V>
384429
extends PTransform<PCollection<KV<K, V>>,
385430
PCollection<KV<K, Iterable<V>>>> {
386-
@Override
387-
public void validate(PCollection<KV<K, V>> input) {
388-
WindowingStrategy<?, ?> windowingStrategy = input.getWindowingStrategy();
389-
if (windowingStrategy.getWindowFn() instanceof GlobalWindows
390-
&& windowingStrategy.getTrigger().getSpec() instanceof DefaultTrigger
391-
&& input.isBounded() != IsBounded.BOUNDED) {
392-
throw new IllegalStateException("Non-bounded PCollection cannot be "
393-
+ "processed with GlobalWindow and DefaultTrigger for GroupByKey."
394-
+ "Use Window.into transform prior to GroupByKey.");
395-
}
396-
// Verify that the input Coder<KV<K, V>> is a KvCoder<K, V>, and that
397-
// the key coder is deterministic.
398-
Coder<K> keyCoder = getKeyCoder(input.getCoder());
399-
try {
400-
keyCoder.verifyDeterministic();
401-
} catch (NonDeterministicException e) {
402-
throw new IllegalStateException(
403-
"the keyCoder of a GroupByKey must be deterministic", e);
404-
}
405-
}
406431

407432
@SuppressWarnings({"rawtypes", "unchecked"})
408433
@Override
409434
public PCollection<KV<K, Iterable<V>>> apply(PCollection<KV<K, V>> input) {
410-
WindowingStrategy<?, ?> oldWindowingStrategy = input.getWindowingStrategy();
411-
WindowFn<?, ?> newWindowFn = oldWindowingStrategy.getWindowFn();
412-
if (!newWindowFn.isNonMerging()) {
413-
// Prevent merging windows again, without explicit user
414-
// involvement, e.g., by Window.into() or Window.remerge().
415-
newWindowFn = new InvalidWindows(
416-
"WindowFn has already been consumed by previous GroupByKey", newWindowFn);
417-
}
418-
419-
// We also switch to the continuation trigger associated with the current trigger.
420-
WindowingStrategy<?, ?> newWindowingStrategy = oldWindowingStrategy
421-
.withWindowFn(newWindowFn)
422-
.withTrigger(oldWindowingStrategy.getTrigger().getSpec().getContinuationTrigger());
423-
424435
return PCollection.<KV<K, Iterable<V>>>createPrimitiveOutputInternal(
425-
input.getPipeline(), newWindowingStrategy, input.isBounded());
436+
input.getPipeline(), input.getWindowingStrategy(), input.isBounded());
426437
}
427438

428439
/**

‎sdk/src/main/java/com/google/cloud/dataflow/sdk/values/PCollection.java‎

Lines changed: 3 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -181,7 +181,6 @@ public IsBounded isBounded() {
181181
return isBounded;
182182
}
183183

184-
185184
/////////////////////////////////////////////////////////////////////////////
186185
// Internal details below here.
187186

@@ -223,8 +222,10 @@ public PCollection<T> setWindowingStrategyInternal(WindowingStrategy<?, ?> windo
223222

224223
/**
225224
* Sets the {@link PCollection.IsBounded} of this {@code PCollection}.
225+
*
226+
* <p> For use by internal transformations only.
226227
*/
227-
private PCollection<T> setIsBoundedInternal(IsBounded isBounded) {
228+
public PCollection<T> setIsBoundedInternal(IsBounded isBounded) {
228229
this.isBounded = isBounded;
229230
return this;
230231
}

‎sdk/src/test/java/com/google/cloud/dataflow/sdk/transforms/GroupByKeyTest.java‎

Lines changed: 100 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -26,14 +26,22 @@
2626
import com.google.cloud.dataflow.sdk.coders.KvCoder;
2727
import com.google.cloud.dataflow.sdk.coders.MapCoder;
2828
import com.google.cloud.dataflow.sdk.coders.StringUtf8Coder;
29+
import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
30+
import com.google.cloud.dataflow.sdk.options.DirectPipelineOptions;
31+
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
32+
import com.google.cloud.dataflow.sdk.runners.DataflowPipelineRunner;
33+
import com.google.cloud.dataflow.sdk.runners.DirectPipelineRunner;
2934
import com.google.cloud.dataflow.sdk.testing.DataflowAssert;
3035
import com.google.cloud.dataflow.sdk.testing.RunnableOnService;
3136
import com.google.cloud.dataflow.sdk.testing.TestPipeline;
3237
import com.google.cloud.dataflow.sdk.transforms.windowing.FixedWindows;
3338
import com.google.cloud.dataflow.sdk.transforms.windowing.InvalidWindows;
3439
import com.google.cloud.dataflow.sdk.transforms.windowing.Sessions;
3540
import com.google.cloud.dataflow.sdk.transforms.windowing.Window;
41+
import com.google.cloud.dataflow.sdk.util.NoopPathValidator;
42+
import com.google.cloud.dataflow.sdk.util.WindowingStrategy;
3643
import com.google.cloud.dataflow.sdk.values.KV;
44+
import com.google.cloud.dataflow.sdk.values.PBegin;
3745
import com.google.cloud.dataflow.sdk.values.PCollection;
3846

3947
import org.joda.time.Duration;
@@ -57,7 +65,7 @@
5765
public class GroupByKeyTest {
5866

5967
@Rule
60-
public ExpectedException expectedEx = ExpectedException.none();
68+
public ExpectedException thrown = ExpectedException.none();
6169

6270
@Test
6371
@Category(RunnableOnService.class)
@@ -167,8 +175,6 @@ public void testGroupByKeyEmpty() {
167175

168176
@Test
169177
public void testGroupByKeyNonDeterministic() throws Exception {
170-
expectedEx.expect(IllegalStateException.class);
171-
expectedEx.expectMessage("must be deterministic");
172178

173179
List<KV<Map<String, String>, Integer>> ungroupedPairs = Arrays.asList();
174180

@@ -180,9 +186,9 @@ public void testGroupByKeyNonDeterministic() throws Exception {
180186
KvCoder.of(MapCoder.of(StringUtf8Coder.of(), StringUtf8Coder.of()),
181187
BigEndianIntegerCoder.of())));
182188

189+
thrown.expect(IllegalStateException.class);
190+
thrown.expectMessage("must be deterministic");
183191
input.apply(GroupByKey.<Map<String, String>, Integer>create());
184-
185-
p.run();
186192
}
187193

188194
@Test
@@ -230,9 +236,30 @@ public void testWindowFnInvalidation() {
230236
Duration.standardMinutes(1)))));
231237
}
232238

239+
/**
240+
* Create a test pipeline that uses the {@link DataflowPipelineRunner} so that {@link GroupByKey}
241+
* is not expanded. This is used for verifying that even without expansion the proper errors show
242+
* up.
243+
*/
244+
private Pipeline createTestServiceRunner() {
245+
DataflowPipelineOptions options = PipelineOptionsFactory.as(DataflowPipelineOptions.class);
246+
options.setRunner(DataflowPipelineRunner.class);
247+
options.setProject("someproject");
248+
options.setStagingLocation("gs://staging");
249+
options.setPathValidatorClass(NoopPathValidator.class);
250+
options.setDataflowClient(null);
251+
return Pipeline.create(options);
252+
}
253+
254+
private Pipeline createTestDirectRunner() {
255+
DirectPipelineOptions options = PipelineOptionsFactory.as(DirectPipelineOptions.class);
256+
options.setRunner(DirectPipelineRunner.class);
257+
return Pipeline.create(options);
258+
}
259+
233260
@Test
234-
public void testInvalidWindows() {
235-
Pipeline p = TestPipeline.create();
261+
public void testInvalidWindowsDirect() {
262+
Pipeline p = createTestDirectRunner();
236263

237264
List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
238265

@@ -242,15 +269,30 @@ public void testInvalidWindows() {
242269
.apply(Window.<KV<String, Integer>>into(
243270
Sessions.withGapDuration(Duration.standardMinutes(1))));
244271

245-
try {
246-
input
247-
.apply("GroupByKey", GroupByKey.<String, Integer>create())
248-
.apply("GroupByKeyAgain", GroupByKey.<String, Iterable<Integer>>create());
249-
Assert.fail("Exception should have been thrown");
250-
} catch (IllegalStateException e) {
251-
Assert.assertTrue(e.getMessage().startsWith(
252-
"GroupByKey must have a valid Window merge function."));
253-
}
272+
thrown.expect(IllegalStateException.class);
273+
thrown.expectMessage("GroupByKey must have a valid Window merge function");
274+
input
275+
.apply("GroupByKey", GroupByKey.<String, Integer>create())
276+
.apply("GroupByKeyAgain", GroupByKey.<String, Iterable<Integer>>create());
277+
}
278+
279+
@Test
280+
public void testInvalidWindowsService() {
281+
Pipeline p = createTestServiceRunner();
282+
283+
List<KV<String, Integer>> ungroupedPairs = Arrays.asList();
284+
285+
PCollection<KV<String, Integer>> input =
286+
p.apply(Create.of(ungroupedPairs)
287+
.withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of())))
288+
.apply(Window.<KV<String, Integer>>into(
289+
Sessions.withGapDuration(Duration.standardMinutes(1))));
290+
291+
thrown.expect(IllegalStateException.class);
292+
thrown.expectMessage("GroupByKey must have a valid Window merge function");
293+
input
294+
.apply("GroupByKey", GroupByKey.<String, Integer>create())
295+
.apply("GroupByKeyAgain", GroupByKey.<String, Iterable<Integer>>create());
254296
}
255297

256298
@Test
@@ -278,6 +320,48 @@ public void testRemerge() {
278320
Sessions.withGapDuration(Duration.standardMinutes(1))));
279321
}
280322

323+
@Test
324+
public void testGroupByKeyDirectUnbounded() {
325+
Pipeline p = createTestDirectRunner();
326+
327+
PCollection<KV<String, Integer>> input = p
328+
.apply(new PTransform<PBegin, PCollection<KV<String, Integer>>>() {
329+
@Override
330+
public PCollection<KV<String, Integer>> apply(PBegin input) {
331+
return PCollection.createPrimitiveOutputInternal(input.getPipeline(),
332+
WindowingStrategy.globalDefault(), PCollection.IsBounded.UNBOUNDED);
333+
}
334+
});
335+
336+
thrown.expect(IllegalStateException.class);
337+
thrown.expectMessage(
338+
"GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without "
339+
+ "a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.");
340+
341+
input.apply("GroupByKey", GroupByKey.<String, Integer>create());
342+
}
343+
344+
@Test
345+
public void testGroupByKeyServiceUnbounded() {
346+
Pipeline p = createTestServiceRunner();
347+
348+
PCollection<KV<String, Integer>> input = p
349+
.apply(new PTransform<PBegin, PCollection<KV<String, Integer>>>() {
350+
@Override
351+
public PCollection<KV<String, Integer>> apply(PBegin input) {
352+
return PCollection.createPrimitiveOutputInternal(input.getPipeline(),
353+
WindowingStrategy.globalDefault(), PCollection.IsBounded.UNBOUNDED);
354+
}
355+
});
356+
357+
thrown.expect(IllegalStateException.class);
358+
thrown.expectMessage(
359+
"GroupByKey cannot be applied to non-bounded PCollection in the GlobalWindow without "
360+
+ "a trigger. Use a Window.into or Window.triggering transform prior to GroupByKey.");
361+
362+
input.apply("GroupByKey", GroupByKey.<String, Integer>create());
363+
}
364+
281365
@Test
282366
public void testGroupByKeyGetName() {
283367
Assert.assertEquals("GroupByKey", GroupByKey.<String, Integer>create().getName());

0 commit comments

Comments
 (0)
Back | FazBrowse Home | New Git URL