[FLINK-14382][examples] Adding multiple --input support to WordCount

spotify · Nov 15, 2019 · 304736d · 304736d
1 parent 468415b
commit 304736d
Show file tree

Hide file tree

Showing 2 changed files with 26 additions and 10 deletions.
diff --git a/...link-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java b/...link-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java
@@ -22,9 +22,10 @@
 import org.apache.flink.api.java.DataSet;
 import org.apache.flink.api.java.ExecutionEnvironment;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.api.java.utils.MultipleParameterTool;
 import org.apache.flink.examples.java.wordcount.util.WordCountData;
 import org.apache.flink.util.Collector;
+import org.apache.flink.util.Preconditions;
 
 /**
  * Implements the "WordCount" program that computes a simple word occurrence histogram
@@ -51,7 +52,7 @@ public class WordCount {
 
  public static void main(String[] args) throws Exception {
 
- final ParameterTool params = ParameterTool.fromArgs(args);
+ final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
 
  // set up the execution environment
  final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
@@ -60,10 +61,17 @@ public static void main(String[] args) throws Exception {
  env.getConfig().setGlobalJobParameters(params);
 
  // get input data
- DataSet<String> text;
+ DataSet<String> text = null;
  if (params.has("input")) {
- // read the text file from given input path
- text = env.readTextFile(params.get("input"));
+ // union all the inputs from text files
+ for (String input : params.getMultiParameterRequired("input")) {
+ if (text == null) {
+ text = env.readTextFile(input);
+ } else {
+ text = text.union(env.readTextFile(input));
+ }
+ }
+ Preconditions.checkNotNull(text, "Input DataSet should not be null.");
  } else {
  // get default test text data
  System.out.println("Executing WordCount example with default input data set.");

diff --git a/...ples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java b/...ples-streaming/src/main/java/org/apache/flink/streaming/examples/wordcount/WordCount.java
@@ -19,11 +19,12 @@
 
 import org.apache.flink.api.common.functions.FlatMapFunction;
 import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.api.java.utils.ParameterTool;
+import org.apache.flink.api.java.utils.MultipleParameterTool;
 import org.apache.flink.streaming.api.datastream.DataStream;
 import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
 import org.apache.flink.streaming.examples.wordcount.util.WordCountData;
 import org.apache.flink.util.Collector;
+import org.apache.flink.util.Preconditions;
 
 /**
  * Implements the "WordCount" program that computes a simple word occurrence
@@ -51,7 +52,7 @@ public class WordCount {
  public static void main(String[] args) throws Exception {
 
  // Checking input parameters
- final ParameterTool params = ParameterTool.fromArgs(args);
+ final MultipleParameterTool params = MultipleParameterTool.fromArgs(args);
 
  // set up the execution environment
  final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
@@ -60,10 +61,17 @@ public static void main(String[] args) throws Exception {
  env.getConfig().setGlobalJobParameters(params);
 
  // get input data
- DataStream<String> text;
+ DataStream<String> text = null;
  if (params.has("input")) {
- // read the text file from given input path
- text = env.readTextFile(params.get("input"));
+ // union all the inputs from text files
+ for (String input : params.getMultiParameterRequired("input")) {
+ if (text == null) {
+ text = env.readTextFile(input);
+ } else {
+ text = text.union(env.readTextFile(input));
+ }
+ }
+ Preconditions.checkNotNull(text, "Input DataStream should not be null.");
  } else {
  System.out.println("Executing WordCount example with default input data set.");
  System.out.println("Use --input to specify file input.");