@Test public void parseBooks_returnsNgramsCounts() { Pipeline p = TestPipeline.create(); PCollection<String> input = p.apply(Create.of(testFile)); PCollection<KV<String, Integer>> output = LoadBooks.applyPipelineToParseBooks(input); DataflowAssert.that(output) .containsInAnyOrder( KV.of("despatch when art", 10), KV.of("despatch when came", 10), KV.of("despatch when published", 12), KV.of("despatch where was", 10), KV.of("despatch which made", 45), KV.of("despatch which addressed", 12 + 46), KV.of("despatch which admitted", 13), KV.of("despatch which allow", 14), KV.of("despatch which announced", 50), KV.of("despatch which answer", 32)); }
static PCollection<KV<String, Integer>> applyPipelineToParseBooks(PCollection<String> p) { return p.apply( FlatMapElements.via((String doc) -> Splitter.on("\n").split(doc)) .withOutputType(new TypeDescriptor<String>() {})) .apply(Filter.byPredicate((String line) -> !line.isEmpty())) .apply( MapElements.via( (String line) -> { List<String> elements = Splitter.on("\t").splitToList(line); String ngram = elements.get(1); String key = getKey(ngram); Integer count = Integer.parseInt(elements.get(2)); return KV.<String, Integer>of(key, count); }) .withOutputType(new TypeDescriptor<KV<String, Integer>>() {})) .apply(Sum.integersPerKey()); }
LoadBooks { static PCollection<KV<String, Integer>> applyPipelineToParseBooks(PCollection<String> p) { return p.apply( FlatMapElements.via((String doc) -> Splitter.on("\n").split(doc)) .withOutputType(new TypeDescriptor<String>() {})) .apply(Filter.byPredicate((String line) -> !line.isEmpty())) .apply( MapElements.via( (String line) -> { List<String> elements = Splitter.on("\t").splitToList(line); String ngram = elements.get(1); String key = getKey(ngram); Integer count = Integer.parseInt(elements.get(2)); return KV.<String, Integer>of(key, count); }) .withOutputType(new TypeDescriptor<KV<String, Integer>>() {})) .apply(Sum.integersPerKey()); } }
LoadBooks { static PCollection<KV<String, Integer>> applyPipelineToParseBooks(PCollection<String> p) { return p.apply( FlatMapElements.via((String doc) -> Splitter.on("\n").split(doc)) .withOutputType(new TypeDescriptor<String>() {})) .apply(Filter.byPredicate((String line) -> !line.isEmpty())) .apply( MapElements.via( (String line) -> { List<String> elements = Splitter.on("\t").splitToList(line); String ngram = elements.get(1); String key = getKey(ngram); Integer count = Integer.parseInt(elements.get(2)); return KV.<String, Integer>of(key, count); }) .withOutputType(new TypeDescriptor<KV<String, Integer>>() {})) .apply(Sum.integersPerKey()); }  }
LoadBooks { static PCollection<KV<String, Integer>> applyPipelineToParseBooks(PCollection<String> p) { return p.apply( FlatMapElements.via((String doc) -> Splitter.on("\n").split(doc)) .withOutputType(new TypeDescriptor<String>() {})) .apply(Filter.byPredicate((String line) -> !line.isEmpty())) .apply( MapElements.via( (String line) -> { List<String> elements = Splitter.on("\t").splitToList(line); String ngram = elements.get(1); String key = getKey(ngram); Integer count = Integer.parseInt(elements.get(2)); return KV.<String, Integer>of(key, count); }) .withOutputType(new TypeDescriptor<KV<String, Integer>>() {})) .apply(Sum.integersPerKey()); }  static void main(String[] args); }
LoadBooks { static PCollection<KV<String, Integer>> applyPipelineToParseBooks(PCollection<String> p) { return p.apply( FlatMapElements.via((String doc) -> Splitter.on("\n").split(doc)) .withOutputType(new TypeDescriptor<String>() {})) .apply(Filter.byPredicate((String line) -> !line.isEmpty())) .apply( MapElements.via( (String line) -> { List<String> elements = Splitter.on("\t").splitToList(line); String ngram = elements.get(1); String key = getKey(ngram); Integer count = Integer.parseInt(elements.get(2)); return KV.<String, Integer>of(key, count); }) .withOutputType(new TypeDescriptor<KV<String, Integer>>() {})) .apply(Sum.integersPerKey()); }  static void main(String[] args);  }