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

Adds streaming Dataflow examples. · sysalexis/DataflowJavaSDK@1d32572 · GitHub

  1. sysalexis
  2. DataflowJavaSDK

Repository navigation

Commit 1d32572

Browse files
millsd
authored and
davorbonaci
committed
Adds streaming Dataflow examples.
[] ------------- Created by MOE: http://code.google.com/p/moe-java MOE_MIGRATED_REVID=85730302
1 parent 6e8cfbc commit 1d32572

3 files changed

Lines changed: 408 additions & 0 deletions

File tree

‎examples/src/main/java/com/google/cloud/dataflow/examples/PubsubFileInjector.java‎

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
/*
2+
* Copyright (C) 2014 Google Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package com.google.cloud.dataflow.examples;
18+
19+
import com.google.api.services.pubsub.Pubsub;
20+
import com.google.api.services.pubsub.model.PublishRequest;
21+
import com.google.api.services.pubsub.model.PubsubMessage;
22+
import com.google.cloud.dataflow.sdk.Pipeline;
23+
import com.google.cloud.dataflow.sdk.io.TextIO;
24+
import com.google.cloud.dataflow.sdk.options.Description;
25+
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
26+
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
27+
import com.google.cloud.dataflow.sdk.options.StreamingOptions;
28+
import com.google.cloud.dataflow.sdk.options.Validation;
29+
import com.google.cloud.dataflow.sdk.transforms.DoFn;
30+
import com.google.cloud.dataflow.sdk.transforms.RateLimiting;
31+
import com.google.cloud.dataflow.sdk.util.Transport;
32+
33+
import java.io.IOException;
34+
35+
/**
36+
* A batch Dataflow pipeline for injecting a set of GCS files into
37+
* a PubSub topic line by line.
38+
*
39+
* <p> This is useful for testing streaming
40+
* pipelines. Note that since batch pipelines might retry chunks, this
41+
* does _not_ guarantee exactly-once injection of file data. Some lines may
42+
* be published multiple times.
43+
* </p>
44+
*/
45+
public class PubsubFileInjector {
46+
47+
/** A DoFn that publishes lines to Google Cloud PubSub. */
48+
static class Publish extends DoFn<String, Void> {
49+
private String outputTopic;
50+
public transient Pubsub pubsub;
51+
52+
Publish(String outputTopic) {
53+
this.outputTopic = outputTopic;
54+
}
55+
56+
@Override
57+
public void startBundle(Context context) {
58+
StreamingOptions options =
59+
context.getPipelineOptions().as(StreamingOptions.class);
60+
this.pubsub = Transport.newPubsubClient(options).build();
61+
}
62+
63+
@Override
64+
public void processElement(ProcessContext c) throws IOException {
65+
PubsubMessage pubsubMessage = new PubsubMessage();
66+
pubsubMessage.encodeData(c.element().getBytes());
67+
PublishRequest publishRequest = new PublishRequest();
68+
publishRequest.setTopic(outputTopic).setMessage(pubsubMessage);
69+
this.pubsub.topics().publish(publishRequest).execute();
70+
}
71+
}
72+
73+
/**
74+
* Command line parameter options.
75+
*/
76+
private interface PubsubFileInjectorOptions extends PipelineOptions {
77+
@Description("GCS location of files.")
78+
@Validation.Required
79+
String getInput();
80+
void setInput(String value);
81+
82+
@Description("Topic to publish on.")
83+
@Validation.Required
84+
String getOutputTopic();
85+
void setOutputTopic(String value);
86+
}
87+
88+
/**
89+
* Sets up and starts streaming pipeline.
90+
*/
91+
public static void main(String[] args) {
92+
PubsubFileInjectorOptions options = PipelineOptionsFactory.fromArgs(args)
93+
.withValidation()
94+
.as(PubsubFileInjectorOptions.class);
95+
96+
Pipeline pipeline = Pipeline.create(options);
97+
98+
pipeline
99+
.apply(TextIO.Read.from(options.getInput()))
100+
.apply(RateLimiting.perWorker(new Publish(options.getOutputTopic()))
101+
.withMaxParallelism(20));
102+
103+
pipeline.run();
104+
}
105+
}

‎examples/src/main/java/com/google/cloud/dataflow/examples/StreamingWordExtract.java‎

Lines changed: 141 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,141 @@
1+
/*
2+
* Copyright (C) 2014 Google Inc.
3+
*
4+
* Licensed under the Apache License, Version 2.0 (the "License"); you may not
5+
* use this file except in compliance with the License. You may obtain a copy of
6+
* the License at
7+
*
8+
* http://www.apache.org/licenses/LICENSE-2.0
9+
*
10+
* Unless required by applicable law or agreed to in writing, software
11+
* distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
12+
* WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
13+
* License for the specific language governing permissions and limitations under
14+
* the License.
15+
*/
16+
17+
package com.google.cloud.dataflow.examples;
18+
19+
import com.google.api.services.bigquery.model.TableFieldSchema;
20+
import com.google.api.services.bigquery.model.TableRow;
21+
import com.google.api.services.bigquery.model.TableSchema;
22+
import com.google.cloud.dataflow.sdk.Pipeline;
23+
import com.google.cloud.dataflow.sdk.io.BigQueryIO;
24+
import com.google.cloud.dataflow.sdk.io.PubsubIO;
25+
import com.google.cloud.dataflow.sdk.options.DataflowPipelineOptions;
26+
import com.google.cloud.dataflow.sdk.options.Description;
27+
import com.google.cloud.dataflow.sdk.options.PipelineOptions;
28+
import com.google.cloud.dataflow.sdk.options.PipelineOptionsFactory;
29+
import com.google.cloud.dataflow.sdk.options.Validation;
30+
import com.google.cloud.dataflow.sdk.transforms.DoFn;
31+
import com.google.cloud.dataflow.sdk.transforms.ParDo;
32+
33+
import java.util.ArrayList;
34+
35+
/**
36+
* A streaming Dataflow Example using BigQuery output.
37+
*
38+
* <p> This pipeline example lines of text from a PubSub topic, splits each line
39+
* into individual words, capitalizes those words, and writes the output to
40+
* a BigQuery table. </p>
41+
*
42+
* <p> To run this example using the Dataflow service, you must provide an input
43+
* pubsub topic and an output BigQuery table, using the {@literal --inputTopic}
44+
* {@literal --dataset} and {@literal --table} options. Since this is a streaming
45+
* pipeline that never completes, select the non-blocking pipeline runner
46+
* {@literal --runner=DataflowPipelineRunner}.
47+
*/
48+
public class StreamingWordExtract {
49+
50+
/** A DoFn that tokenizes lines of text into individual words. */
51+
static class ExtractWords extends DoFn<String, String> {
52+
@Override
53+
public void processElement(ProcessContext c) {
54+
String[] words = c.element().split("[^a-zA-Z']+");
55+
for (String word : words) {
56+
if (!word.isEmpty()) {
57+
c.output(word);
58+
}
59+
}
60+
}
61+
}
62+
63+
/** A DoFn that uppercases a word. */
64+
static class Uppercase extends DoFn<String, String> {
65+
@Override
66+
public void processElement(ProcessContext c) {
67+
c.output(c.element().toUpperCase());
68+
}
69+
}
70+
71+
/**
72+
* Converts strings into BigQuery rows.
73+
*/
74+
static class StringToRowConverter extends DoFn<String, TableRow> {
75+
76+
/**
77+
* In this example, put the whole string into single BigQuery field.
78+
*/
79+
@Override
80+
public void processElement(ProcessContext c) {
81+
c.output(new TableRow().set("string_field", c.element()));
82+
}
83+
84+
static TableSchema getSchema() {
85+
return new TableSchema().setFields(new ArrayList<TableFieldSchema>() {
86+
// Compose the list of TableFieldSchema from tableSchema.
87+
{
88+
add(new TableFieldSchema().setName("string_field").setType("STRING"));
89+
}
90+
});
91+
}
92+
}
93+
94+
/**
95+
* Command line parameter options.
96+
*/
97+
private interface StreamingWordExtractOptions extends PipelineOptions {
98+
@Description("Input Pubsub topic")
99+
@Validation.Required
100+
String getInputTopic();
101+
void setInputTopic(String value);
102+
103+
@Description("BigQuery dataset name")
104+
@Validation.Required
105+
String getDataset();
106+
void setDataset(String value);
107+
108+
@Description("BigQuery table name")
109+
@Validation.Required
110+
String getTable();
111+
void setTable(String value);
112+
}
113+
114+
/**
115+
* Sets up and starts streaming pipeline.
116+
*/
117+
public static void main(String[] args) {
118+
StreamingWordExtractOptions options = PipelineOptionsFactory.fromArgs(args)
119+
.withValidation()
120+
.as(StreamingWordExtractOptions.class);
121+
DataflowPipelineOptions dataflowOptions = options.as(DataflowPipelineOptions.class);
122+
dataflowOptions.setStreaming(true);
123+
124+
Pipeline pipeline = Pipeline.create(options);
125+
126+
String tableSpec = new StringBuilder()
127+
.append(dataflowOptions.getProject()).append(":")
128+
.append(options.getDataset()).append(".")
129+
.append(options.getTable())
130+
.toString();
131+
pipeline
132+
.apply(PubsubIO.Read.topic(options.getInputTopic()))
133+
.apply(ParDo.of(new ExtractWords()))
134+
.apply(ParDo.of(new Uppercase()))
135+
.apply(ParDo.of(new StringToRowConverter()))
136+
.apply(BigQueryIO.Write.to(tableSpec)
137+
.withSchema(StringToRowConverter.getSchema()));
138+
139+
pipeline.run();
140+
}
141+
}

0 commit comments

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