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

Add v1beta3 version of datastore Source/Sink. (#328) · walternate/DataflowJavaSDK@0410aaf · GitHub

  1. walternate
  2. DataflowJavaSDK

Repository navigation

Commit 0410aaf

Browse files
vikkyrk
authored and
dhalperi
committed
Add v1beta3 version of datastore Source/Sink. (GoogleCloudPlatform#328)
* Add v1beta3 version of datastore Source/Sink. * Deprecate old DatastoreIO classes * getProject to getDataset
1 parent 088f6f1 commit 0410aaf

13 files changed

Lines changed: 1779 additions & 108 deletions

File tree

‎examples/pom.xml‎

Lines changed: 12 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -459,6 +459,18 @@
459459
</exclusions>
460460
</dependency>
461461

462+
<dependency>
463+
<groupId>com.google.cloud.datastore</groupId>
464+
<artifactId>datastore-v1beta3-proto-client</artifactId>
465+
<version>${datastore.client.version}</version>
466+
</dependency>
467+
468+
<dependency>
469+
<groupId>com.google.cloud.datastore</groupId>
470+
<artifactId>datastore-v1beta3-protos</artifactId>
471+
<version>${datastore.proto.version}</version>
472+
</dependency>
473+
462474
<dependency>
463475
<groupId>com.google.guava</groupId>
464476
<artifactId>guava</artifactId>

‎examples/src/main/java/com/google/cloud/dataflow/examples/complete/AutoComplete.java‎

Lines changed: 81 additions & 78 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,12 @@
1616

1717
package com.google.cloud.dataflow.examples.complete;
1818

19+
import static com.google.datastore.v1beta3.client.DatastoreHelper.makeValue;
20+
1921
import com.google.api.services.bigquery.model.TableFieldSchema;
2022
import com.google.api.services.bigquery.model.TableReference;
2123
import com.google.api.services.bigquery.model.TableRow;
2224
import com.google.api.services.bigquery.model.TableSchema;
23-
import com.google.api.services.datastore.DatastoreV1.Entity;
24-
import com.google.api.services.datastore.DatastoreV1.Key;
25-
import com.google.api.services.datastore.DatastoreV1.Value;
26-
import com.google.api.services.datastore.client.DatastoreHelper;
2725
import com.google.cloud.dataflow.examples.common.DataflowExampleUtils;
2826
import com.google.cloud.dataflow.examples.common.ExampleBigQueryTableOptions;
2927
import com.google.cloud.dataflow.examples.common.ExamplePubsubTopicOptions;
@@ -32,9 +30,9 @@
3230
import com.google.cloud.dataflow.sdk.coders.AvroCoder;
3331
import com.google.cloud.dataflow.sdk.coders.DefaultCoder;
3432
import com.google.cloud.dataflow.sdk.io.BigQueryIO;
35-
import com.google.cloud.dataflow.sdk.io.DatastoreIO;
3633
import com.google.cloud.dataflow.sdk.io.PubsubIO;
3734
import com.google.cloud.dataflow.sdk.io.TextIO;
35+
import com.google.cloud.dataflow.sdk.io.datastore.DatastoreIO;
3836
import com.google.cloud.dataflow.sdk.options.Default;
3937
import com.google.cloud.dataflow.sdk.options.Description;
4038
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
@@ -60,12 +58,18 @@
6058
import com.google.cloud.dataflow.sdk.values.PCollectionList;
6159
import com.google.common.base.MoreObjects;
6260
import com.google.common.base.Preconditions;
61+
import com.google.datastore.v1beta3.Entity;
62+
import com.google.datastore.v1beta3.Key;
63+
import com.google.datastore.v1beta3.Value;
64+
import com.google.datastore.v1beta3.client.DatastoreHelper;
6365

6466
import org.joda.time.Duration;
6567

6668
import java.io.IOException;
6769
import java.util.ArrayList;
70+
import java.util.HashMap;
6871
import java.util.List;
72+
import java.util.Map;
6973
import java.util.regex.Matcher;
7074
import java.util.regex.Pattern;
7175

@@ -121,26 +125,26 @@ public static ComputeTopCompletions top(int candidatesPerPrefix, boolean recursi
121125
@Override
122126
public PCollection<KV<String, List<CompletionCandidate>>> apply(PCollection<String> input) {
123127
PCollection<CompletionCandidate> candidates = input
124-
// First count how often each token appears.
125-
.apply(new Count.PerElement<String>())
126-
127-
// Map the KV outputs of Count into our own CompletionCandiate class.
128-
.apply(ParDo.named("CreateCompletionCandidates").of(
129-
new DoFn<KV<String, Long>, CompletionCandidate>() {
130-
@Override
131-
public void processElement(ProcessContext c) {
132-
c.output(new CompletionCandidate(c.element().getKey(), c.element().getValue()));
133-
}
134-
}));
128+
// First count how often each token appears.
129+
.apply(new Count.PerElement<String>())
130+
131+
// Map the KV outputs of Count into our own CompletionCandiate class.
132+
.apply(ParDo.named("CreateCompletionCandidates").of(
133+
new DoFn<KV<String, Long>, CompletionCandidate>() {
134+
@Override
135+
public void processElement(ProcessContext c) {
136+
c.output(new CompletionCandidate(c.element().getKey(), c.element().getValue()));
137+
}
138+
}));
135139

136140
// Compute the top via either a flat or recursive algorithm.
137141
if (recursive) {
138142
return candidates
139-
.apply(new ComputeTopRecursive(candidatesPerPrefix, 1))
140-
.apply(Flatten.<KV<String, List<CompletionCandidate>>>pCollections());
143+
.apply(new ComputeTopRecursive(candidatesPerPrefix, 1))
144+
.apply(Flatten.<KV<String, List<CompletionCandidate>>>pCollections());
141145
} else {
142146
return candidates
143-
.apply(new ComputeTopFlat(candidatesPerPrefix, 1));
147+
.apply(new ComputeTopFlat(candidatesPerPrefix, 1));
144148
}
145149
}
146150
}
@@ -150,7 +154,7 @@ public void processElement(ProcessContext c) {
150154
*/
151155
private static class ComputeTopFlat
152156
extends PTransform<PCollection<CompletionCandidate>,
153-
PCollection<KV<String, List<CompletionCandidate>>>> {
157+
PCollection<KV<String, List<CompletionCandidate>>>> {
154158
private final int candidatesPerPrefix;
155159
private final int minPrefix;
156160

@@ -163,12 +167,12 @@ public ComputeTopFlat(int candidatesPerPrefix, int minPrefix) {
163167
public PCollection<KV<String, List<CompletionCandidate>>> apply(
164168
PCollection<CompletionCandidate> input) {
165169
return input
166-
// For each completion candidate, map it to all prefixes.
167-
.apply(ParDo.of(new AllPrefixes(minPrefix)))
170+
// For each completion candidate, map it to all prefixes.
171+
.apply(ParDo.of(new AllPrefixes(minPrefix)))
168172

169-
// Find and return the top candiates for each prefix.
170-
.apply(Top.<String, CompletionCandidate>largestPerKey(candidatesPerPrefix)
171-
.withHotKeyFanout(new HotKeyFanout()));
173+
// Find and return the top candiates for each prefix.
174+
.apply(Top.<String, CompletionCandidate>largestPerKey(candidatesPerPrefix)
175+
.withHotKeyFanout(new HotKeyFanout()));
172176
}
173177

174178
private static class HotKeyFanout implements SerializableFunction<String, Integer> {
@@ -188,7 +192,7 @@ public Integer apply(String input) {
188192
*/
189193
private static class ComputeTopRecursive
190194
extends PTransform<PCollection<CompletionCandidate>,
191-
PCollectionList<KV<String, List<CompletionCandidate>>>> {
195+
PCollectionList<KV<String, List<CompletionCandidate>>>> {
192196
private final int candidatesPerPrefix;
193197
private final int minPrefix;
194198

@@ -216,42 +220,42 @@ public void processElement(ProcessContext c) {
216220

217221
@Override
218222
public PCollectionList<KV<String, List<CompletionCandidate>>> apply(
219-
PCollection<CompletionCandidate> input) {
220-
if (minPrefix > 10) {
221-
// Base case, partitioning to return the output in the expected format.
222-
return input
223+
PCollection<CompletionCandidate> input) {
224+
if (minPrefix > 10) {
225+
// Base case, partitioning to return the output in the expected format.
226+
return input
223227
.apply(new ComputeTopFlat(candidatesPerPrefix, minPrefix))
224228
.apply(Partition.of(2, new KeySizePartitionFn()));
225-
} else {
226-
// If a candidate is in the top N for prefix a...b, it must also be in the top
227-
// N for a...bX for every X, which is typlically a much smaller set to consider.
228-
// First, compute the top candidate for prefixes of size at least minPrefix + 1.
229-
PCollectionList<KV<String, List<CompletionCandidate>>> larger = input
229+
} else {
230+
// If a candidate is in the top N for prefix a...b, it must also be in the top
231+
// N for a...bX for every X, which is typlically a much smaller set to consider.
232+
// First, compute the top candidate for prefixes of size at least minPrefix + 1.
233+
PCollectionList<KV<String, List<CompletionCandidate>>> larger = input
230234
.apply(new ComputeTopRecursive(candidatesPerPrefix, minPrefix + 1));
231-
// Consider the top candidates for each prefix of length minPrefix + 1...
232-
PCollection<KV<String, List<CompletionCandidate>>> small =
235+
// Consider the top candidates for each prefix of length minPrefix + 1...
236+
PCollection<KV<String, List<CompletionCandidate>>> small =
233237
PCollectionList
234-
.of(larger.get(1).apply(ParDo.of(new FlattenTops())))
235-
// ...together with those (previously excluded) candidates of length
236-
// exactly minPrefix...
237-
.and(input.apply(Filter.byPredicate(
238-
new SerializableFunction<CompletionCandidate, Boolean>() {
239-
@Override
240-
public Boolean apply(CompletionCandidate c) {
241-
return c.getValue().length() == minPrefix;
242-
}
243-
})))
244-
.apply("FlattenSmall", Flatten.<CompletionCandidate>pCollections())
245-
// ...set the key to be the minPrefix-length prefix...
246-
.apply(ParDo.of(new AllPrefixes(minPrefix, minPrefix)))
247-
// ...and (re)apply the Top operator to all of them together.
248-
.apply(Top.<String, CompletionCandidate>largestPerKey(candidatesPerPrefix));
249-
250-
PCollection<KV<String, List<CompletionCandidate>>> flattenLarger = larger
251-
.apply("FlattenLarge", Flatten.<KV<String, List<CompletionCandidate>>>pCollections());
252-
253-
return PCollectionList.of(flattenLarger).and(small);
254-
}
238+
.of(larger.get(1).apply(ParDo.of(new FlattenTops())))
239+
// ...together with those (previously excluded) candidates of length
240+
// exactly minPrefix...
241+
.and(input.apply(Filter.byPredicate(
242+
new SerializableFunction<CompletionCandidate, Boolean>() {
243+
@Override
244+
public Boolean apply(CompletionCandidate c) {
245+
return c.getValue().length() == minPrefix;
246+
}
247+
})))
248+
.apply("FlattenSmall", Flatten.<CompletionCandidate>pCollections())
249+
// ...set the key to be the minPrefix-length prefix...
250+
.apply(ParDo.of(new AllPrefixes(minPrefix, minPrefix)))
251+
// ...and (re)apply the Top operator to all of them together.
252+
.apply(Top.<String, CompletionCandidate>largestPerKey(candidatesPerPrefix));
253+
254+
PCollection<KV<String, List<CompletionCandidate>>> flattenLarger = larger
255+
.apply("FlattenLarge", Flatten.<KV<String, List<CompletionCandidate>>>pCollections());
256+
257+
return PCollectionList.of(flattenLarger).and(small);
258+
}
255259
}
256260
}
257261

@@ -270,7 +274,7 @@ public AllPrefixes(int minPrefix, int maxPrefix) {
270274
this.maxPrefix = maxPrefix;
271275
}
272276
@Override
273-
public void processElement(ProcessContext c) {
277+
public void processElement(ProcessContext c) {
274278
String word = c.element().value;
275279
for (int i = minPrefix; i <= Math.min(word.length(), maxPrefix); i++) {
276280
c.output(KV.of(word.substring(0, i), c.element()));
@@ -353,12 +357,12 @@ public void processElement(ProcessContext c) {
353357
List<TableRow> completions = new ArrayList<>();
354358
for (CompletionCandidate cc : c.element().getValue()) {
355359
completions.add(new TableRow()
356-
.set("count", cc.getCount())
357-
.set("tag", cc.getValue()));
360+
.set("count", cc.getCount())
361+
.set("tag", cc.getValue()));
358362
}
359363
TableRow row = new TableRow()
360-
.set("prefix", c.element().getKey())
361-
.set("tags", completions);
364+
.set("prefix", c.element().getKey())
365+
.set("tags", completions);
362366
c.output(row);
363367
}
364368

@@ -395,16 +399,15 @@ public void processElement(ProcessContext c) {
395399

396400
entityBuilder.setKey(key);
397401
List<Value> candidates = new ArrayList<>();
402+
Map<String, Value> properties = new HashMap<>();
398403
for (CompletionCandidate tag : c.element().getValue()) {
399404
Entity.Builder tagEntity = Entity.newBuilder();
400-
tagEntity.addProperty(
401-
DatastoreHelper.makeProperty("tag", DatastoreHelper.makeValue(tag.value)));
402-
tagEntity.addProperty(
403-
DatastoreHelper.makeProperty("count", DatastoreHelper.makeValue(tag.count)));
404-
candidates.add(DatastoreHelper.makeValue(tagEntity).setIndexed(false).build());
405+
properties.put("tag", makeValue(tag.value).build());
406+
properties.put("count", makeValue(tag.count).build());
407+
candidates.add(makeValue(tagEntity).build());
405408
}
406-
entityBuilder.addProperty(
407-
DatastoreHelper.makeProperty("candidates", DatastoreHelper.makeValue(candidates)));
409+
properties.put("candidates", makeValue(candidates).build());
410+
entityBuilder.putAllProperties(properties);
408411
c.output(entityBuilder.build());
409412
}
410413
}
@@ -425,7 +428,7 @@ private static interface Options extends ExamplePubsubTopicOptions, ExampleBigQu
425428
Boolean getRecursive();
426429
void setRecursive(Boolean value);
427430

428-
@Description("Dataset entity kind")
431+
@Description("Datastore entity kind")
429432
@Default.String("autocomplete-demo")
430433
String getKind();
431434
void setKind(String value);
@@ -476,16 +479,16 @@ public static void main(String[] args) throws IOException {
476479
// Create the pipeline.
477480
Pipeline p = Pipeline.create(options);
478481
PCollection<KV<String, List<CompletionCandidate>>> toWrite = p
479-
.apply(readSource)
480-
.apply(ParDo.of(new ExtractHashtags()))
481-
.apply(Window.<String>into(windowFn))
482-
.apply(ComputeTopCompletions.top(10, options.getRecursive()));
482+
.apply(readSource)
483+
.apply(ParDo.of(new ExtractHashtags()))
484+
.apply(Window.<String>into(windowFn))
485+
.apply(ComputeTopCompletions.top(10, options.getRecursive()));
483486

484487
if (options.getOutputToDatastore()) {
485488
toWrite
486-
.apply(ParDo.named("FormatForDatastore").of(new FormatForDatastore(options.getKind())))
487-
.apply(DatastoreIO.writeTo(MoreObjects.firstNonNull(
488-
options.getOutputDataset(), options.getProject())));
489+
.apply(ParDo.named("FormatForDatastore").of(new FormatForDatastore(options.getKind())))
490+
.apply(DatastoreIO.v1beta3().write().withProjectId(MoreObjects.firstNonNull(
491+
options.getOutputDataset(), options.getProject())));
489492
}
490493
if (options.getOutputToBigQuery()) {
491494
dataflowUtils.setupBigQueryTable();

0 commit comments

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