You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by rm...@apache.org on 2016/02/23 19:36:03 UTC
[3/3] flink git commit: [FLINK-2021] Rework examples to use
ParameterTool
[FLINK-2021] Rework examples to use ParameterTool
This closes #1581
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/0629e256
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/0629e256
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/0629e256
Branch: refs/heads/master
Commit: 0629e25602eefdc239e8e72d9e3c9c1a5164448e
Parents: c40cfd3
Author: Stefano Baghino <st...@baghino.me>
Authored: Thu Jan 21 12:20:49 2016 +0100
Committer: Robert Metzger <rm...@apache.org>
Committed: Tue Feb 23 19:34:47 2016 +0100
----------------------------------------------------------------------
docs/apis/batch/examples.md | 20 +-
docs/quickstart/run_example_quickstart.md | 6 +-
.../flink/examples/java/clustering/KMeans.java | 243 ++++++++-----------
.../flink/examples/java/distcp/DistCp.java | 47 ++--
.../java/graph/ConnectedComponents.java | 92 +++----
.../examples/java/graph/EnumTriangles.java | 77 ++----
.../flink/examples/java/graph/PageRank.java | 106 ++++----
.../java/graph/TransitiveClosureNaive.java | 72 ++----
.../examples/java/ml/LinearRegression.java | 107 ++------
.../java/ml/util/LinearRegressionData.java | 4 +-
.../relational/EmptyFieldsCountAccumulator.java | 105 ++++----
.../examples/java/relational/TPCHQuery10.java | 80 +++---
.../examples/java/relational/TPCHQuery3.java | 66 ++---
.../java/relational/WebLogAnalysis.java | 74 +++---
.../examples/java/wordcount/WordCount.java | 69 ++----
.../examples/java/wordcount/WordCountPojo.java | 66 ++---
.../examples/scala/clustering/KMeans.scala | 190 +++++++--------
.../scala/graph/ConnectedComponents.scala | 85 +++----
.../examples/scala/graph/DeltaPageRank.scala | 2 +-
.../examples/scala/graph/EnumTriangles.scala | 73 ++----
.../examples/scala/graph/PageRankBasic.scala | 83 +++----
.../scala/graph/TransitiveClosureNaive.scala | 82 +++----
.../examples/scala/ml/LinearRegression.scala | 102 +++-----
.../examples/scala/relational/TPCHQuery10.scala | 87 ++++---
.../examples/scala/relational/TPCHQuery3.scala | 81 ++++---
.../scala/relational/WebLogAnalysis.scala | 96 +++-----
.../examples/scala/wordcount/WordCount.scala | 59 ++---
.../examples/iteration/IterateExample.java | 63 ++---
.../streaming/examples/join/WindowJoin.java | 74 ++----
.../ml/IncrementalLearningSkeleton.java | 38 +--
.../socket/SocketTextStreamWordCount.java | 49 ++--
.../examples/twitter/TwitterStream.java | 69 ++----
.../examples/windowing/SessionWindowing.java | 36 +--
.../examples/windowing/TopSpeedWindowing.java | 47 +---
.../examples/windowing/WindowWordCount.java | 78 ++----
.../examples/wordcount/PojoExample.java | 65 ++---
.../streaming/examples/wordcount/WordCount.java | 65 ++---
.../scala/examples/join/WindowJoin.scala | 71 ++----
.../socket/SocketTextStreamWordCount.scala | 42 +---
.../examples/windowing/TopSpeedWindowing.scala | 60 ++---
.../iteration/IterateExampleITCase.java | 4 +-
.../join/WindowJoinITCase.java | 5 +-
.../ml/IncrementalLearningSkeletonITCase.java | 2 +-
.../socket/SocketTextStreamWordCountITCase.java | 5 +-
.../twitter/TwitterStreamITCase.java | 2 +-
.../windowing/SessionWindowingITCase.java | 2 +-
.../TopSpeedWindowingExampleITCase.java | 4 +-
.../windowing/WindowWordCountITCase.java | 6 +-
.../wordcount/PojoExampleITCase.java | 4 +-
.../wordcount/WordCountITCase.java | 4 +-
.../join/WindowJoinITCase.java | 5 +-
.../socket/SocketTextStreamWordCountITCase.java | 5 +-
.../TopSpeedWindowingExampleITCase.java | 4 +-
.../org/apache/flink/hdfstests/HDFSTest.java | 4 +-
.../ConnectedComponentsITCase.java | 6 +-
.../EnumTriangleBasicITCase.java | 4 +-
.../exampleJavaPrograms/PageRankITCase.java | 16 +-
.../TransitiveClosureITCase.java | 5 +-
.../WebLogAnalysisITCase.java | 6 +-
.../exampleJavaPrograms/WordCountITCase.java | 4 +-
.../ConnectedComponentsITCase.java | 6 +-
.../EnumTriangleITCase.java | 4 +-
.../exampleScalaPrograms/PageRankITCase.java | 14 +-
.../TransitiveClosureITCase.java | 5 +-
.../WebLogAnalysisITCase.java | 6 +-
.../exampleScalaPrograms/WordCountITCase.java | 4 +-
.../jsonplan/DumpCompiledPlanTest.java | 37 ++-
.../jsonplan/JsonJobGraphGenerationTest.java | 28 ++-
.../optimizer/jsonplan/PreviewPlanDumpTest.java | 38 ++-
.../YARNSessionCapacitySchedulerITCase.java | 2 +-
70 files changed, 1265 insertions(+), 1907 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/docs/apis/batch/examples.md
----------------------------------------------------------------------
diff --git a/docs/apis/batch/examples.md b/docs/apis/batch/examples.md
index 78fc07a..3b6ccd4 100644
--- a/docs/apis/batch/examples.md
+++ b/docs/apis/batch/examples.md
@@ -55,7 +55,7 @@ The other examples can be started in a similar way.
Note that many examples run without passing any arguments for them, by using build-in data. To run WordCount with real data, you have to pass the path to the data:
~~~bash
-./bin/flink run ./examples/batch/WordCount.jar /path/to/some/text/data /path/to/result
+./bin/flink run ./examples/batch/WordCount.jar --input /path/to/some/text/data --output /path/to/result
~~~
Note that non-local file systems require a schema prefix, such as `hdfs://`.
@@ -99,7 +99,7 @@ public static class Tokenizer implements FlatMapFunction<String, Tuple2<String,
}
~~~
-The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java "WordCount example" %} implements the above described algorithm with input parameters: `<text input path>, <output path>`. As test data, any text file will do.
+The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/wordcount/WordCount.java "WordCount example" %} implements the above described algorithm with input parameters: `--input <path> --output <path>`. As test data, any text file will do.
</div>
<div data-lang="scala" markdown="1">
@@ -118,7 +118,7 @@ val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
counts.writeAsCsv(outputPath, "\n", " ")
~~~
-The {% gh_link /flink-examples/flink-exampls-batch/src/main/scala/org/apache/flink/examples/scala/wordcount/WordCount.scala "WordCount example" %} implements the above described algorithm with input parameters: `<text input path>, <output path>`. As test data, any text file will do.
+The {% gh_link /flink-examples/flink-exampls-batch/src/main/scala/org/apache/flink/examples/scala/wordcount/WordCount.scala "WordCount example" %} implements the above described algorithm with input parameters: `--input <path> --output <path>`. As test data, any text file will do.
</div>
@@ -176,7 +176,7 @@ public static final class JoinVertexWithEdgesMatch
double rankToDistribute = rank / ((double) neigbors.length);
for (int i = 0; i < neigbors.length; i++) {
- out.collect(new Tuple2<Long, Double>(neigbors[i], rankToDistribute));
+ out.collect(new Tuple2<Long, Double>(neighbors[i], rankToDistribute));
}
}
}
@@ -207,7 +207,7 @@ public static final class EpsilonFilter
~~~
The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java "PageRank program" %} implements the above example.
-It requires the following parameters to run: `<pages input path>, <links input path>, <output path>, <num pages>, <num iterations>`.
+It requires the following parameters to run: `--pages <path> --links <path> --output <path> --numPages <n> --iterations <n>`.
</div>
<div data-lang="scala" markdown="1">
@@ -274,7 +274,7 @@ result.writeAsCsv(outputPath, "\n", " ")
~~~
he {% gh_link /flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/graph/PageRankBasic.scala "PageRank program" %} implements the above example.
-It requires the following parameters to run: `<pages input path>, <links input path>, <output path>, <num pages>, <num iterations>`.
+It requires the following parameters to run: `--pages <path> --links <path> --output <path> --numPages <n> --iterations <n>`.
</div>
</div>
@@ -369,7 +369,7 @@ public static final class ComponentIdFilter
}
~~~
-The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java "ConnectedComponents program" %} implements the above example. It requires the following parameters to run: `<vertex input path>, <edge input path>, <output path> <max num iterations>`.
+The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java "ConnectedComponents program" %} implements the above example. It requires the following parameters to run: `--vertices <path> --edges <path> --output <path> --iterations <n>`.
</div>
<div data-lang="scala" markdown="1">
@@ -412,7 +412,7 @@ verticesWithComponents.writeAsCsv(outputPath, "\n", " ")
~~~
-The {% gh_link /flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala "ConnectedComponents program" %} implements the above example. It requires the following parameters to run: `<vertex input path>, <edge input path>, <output path> <max num iterations>`.
+The {% gh_link /flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/graph/ConnectedComponents.scala "ConnectedComponents program" %} implements the above example. It requires the following parameters to run: `--vertices <path> --edges <path> --output <path> --iterations <n>`.
</div>
</div>
@@ -488,13 +488,13 @@ DataSet<Tuple3<Integer, Integer, Double>> priceSums =
priceSums.writeAsCsv(outputPath);
~~~
-The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java "Relational Query program" %} implements the above query. It requires the following parameters to run: `<orders input path>, <lineitem input path>, <output path>`.
+The {% gh_link /flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java "Relational Query program" %} implements the above query. It requires the following parameters to run: `--orders <path> --lineitem <path> --output <path>`.
</div>
<div data-lang="scala" markdown="1">
Coming soon...
-The {% gh_link /flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/relational/TPCHQuery3.scala "Relational Query program" %} implements the above query. It requires the following parameters to run: `<orders input path>, <lineitem input path>, <output path>`.
+The {% gh_link /flink-examples/flink-examples-batch/src/main/scala/org/apache/flink/examples/scala/relational/TPCHQuery3.scala "Relational Query program" %} implements the above query. It requires the following parameters to run: `--orders <path> --lineitem <path> --output <path>`.
</div>
</div>
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/docs/quickstart/run_example_quickstart.md
----------------------------------------------------------------------
diff --git a/docs/quickstart/run_example_quickstart.md b/docs/quickstart/run_example_quickstart.md
index 3fd2801..7e3d999 100644
--- a/docs/quickstart/run_example_quickstart.md
+++ b/docs/quickstart/run_example_quickstart.md
@@ -100,10 +100,8 @@ The Flink web interface allows to submit Flink programs using a graphical user i
5. Enter the arguments and options in the lower box: <br>
Leave the <i>Entry Class</i> and <i>Parallelism</i> form empty<br>
Enter the following program arguments: <br>
- (KMeans expects the following args: <code><points path> <centers path> <result path> <num iterations></code>
- {% highlight bash %}
- /tmp/kmeans/points /tmp/kmeans/centers /tmp/kmeans/result 10
- {% endhighlight %} <br>
+ (KMeans expects the following args: <code>--points <path> --centroids <path> --output <path> --iterations <n></code>
+ {% highlight bash %}--points /tmp/kmeans/points --centroids /tmp/kmeans/centers --output /tmp/kmeans/result --iterations 10{% endhighlight %}<br>
6. Press <b>Submit</b> to start the job
</div>
</div>
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java
index 1730e2a..b2e9462 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/clustering/KMeans.java
@@ -27,6 +27,7 @@ import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.examples.java.clustering.util.KMeansData;
import org.apache.flink.api.java.DataSet;
@@ -35,23 +36,23 @@ import org.apache.flink.api.java.operators.IterativeDataSet;
/**
* This example implements a basic K-Means clustering algorithm.
- *
+ *
* <p>
* K-Means is an iterative clustering algorithm and works as follows:<br>
* K-Means is given a set of data points to be clustered and an initial set of <i>K</i> cluster centers.
* In each iteration, the algorithm computes the distance of each data point to each cluster center.
* Each point is assigned to the cluster center which is closest to it.
* Subsequently, each cluster center is moved to the center (<i>mean</i>) of all points that have been assigned to it.
- * The moved cluster centers are fed into the next iteration.
- * The algorithm terminates after a fixed number of iterations (as in this implementation)
+ * The moved cluster centers are fed into the next iteration.
+ * The algorithm terminates after a fixed number of iterations (as in this implementation)
* or if cluster centers do not (significantly) move in an iteration.<br>
* This is the Wikipedia entry for the <a href="http://en.wikipedia.org/wiki/K-means_clustering">K-Means Clustering algorithm</a>.
- *
+ *
* <p>
* This implementation works on two-dimensional data points. <br>
- * It computes an assignment of data points to cluster centers, i.e.,
+ * It computes an assignment of data points to cluster centers, i.e.,
* each data point is annotated with the id of the final cluster (center) it belongs to.
- *
+ *
* <p>
* Input files are plain text files and must be formatted as follows:
* <ul>
@@ -61,42 +62,40 @@ import org.apache.flink.api.java.operators.IterativeDataSet;
* <li>Cluster centers are represented by an integer id and a point value.<br>
* For example <code>"1 6.2 3.2\n2 2.9 5.7\n"</code> gives two centers (id=1, x=6.2, y=3.2) and (id=2, x=2.9, y=5.7).
* </ul>
- *
+ *
* <p>
- * Usage: <code>KMeans <points path> <centers path> <result path> <num iterations></code><br>
- * If no parameters are provided, the program is run with default data from {@link org.apache.flink.examples.java.clustering.util.KMeansData} and 10 iterations.
- *
+ * Usage: <code>KMeans --points <path> --centroids <path> --output <path> --iterations <n></code><br>
+ * If no parameters are provided, the program is run with default data from {@link org.apache.flink.examples.java.clustering.util.KMeansData} and 10 iterations.
+ *
* <p>
* This example shows how to use:
* <ul>
* <li>Bulk iterations
* <li>Broadcast variables in bulk iterations
- * <li>Custom Java objects (PoJos)
+ * <li>Custom Java objects (POJOs)
* </ul>
*/
@SuppressWarnings("serial")
public class KMeans {
-
- // *************************************************************************
- // PROGRAM
- // *************************************************************************
-
+
public static void main(String[] args) throws Exception {
-
- if(!parseParameters(args)) {
- return;
- }
-
+
+ // Checking input parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: KMeans --points <path> --centroids <path> --output <path> --iterations <n>");
+
// set up execution environment
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
-
- // get input data
- DataSet<Point> points = getPointDataSet(env);
- DataSet<Centroid> centroids = getCentroidDataSet(env);
-
+ env.getConfig().setGlobalJobParameters(params); // make parameters available in the web interface
+
+ // get input data:
+ // read the points and centroids from the provided paths or fall back to default data
+ DataSet<Point> points = getPointDataSet(params, env);
+ DataSet<Centroid> centroids = getCentroidDataSet(params, env);
+
// set number of bulk iterations for KMeans algorithm
- IterativeDataSet<Centroid> loop = centroids.iterate(numIterations);
-
+ IterativeDataSet<Centroid> loop = centroids.iterate(params.getInt("iterations", 10));
+
DataSet<Centroid> newCentroids = points
// compute closest centroid for each point
.map(new SelectNearestCenter()).withBroadcastSet(loop, "centroids")
@@ -105,119 +104,132 @@ public class KMeans {
.groupBy(0).reduce(new CentroidAccumulator())
// compute new centroids from point counts and coordinate sums
.map(new CentroidAverager());
-
+
// feed new centroids back into next iteration
DataSet<Centroid> finalCentroids = loop.closeWith(newCentroids);
-
+
DataSet<Tuple2<Integer, Point>> clusteredPoints = points
- // assign points to final clusters
- .map(new SelectNearestCenter()).withBroadcastSet(finalCentroids, "centroids");
-
+ // assign points to final clusters
+ .map(new SelectNearestCenter()).withBroadcastSet(finalCentroids, "centroids");
+
// emit result
- if (fileOutput) {
- clusteredPoints.writeAsCsv(outputPath, "\n", " ");
+ if (params.has("output")) {
+ clusteredPoints.writeAsCsv(params.get("output"), "\n", " ");
// since file sinks are lazy, we trigger the execution explicitly
env.execute("KMeans Example");
- }
- else {
+ } else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
clusteredPoints.print();
}
}
-
+
+ // *************************************************************************
+ // DATA SOURCE READING (POINTS AND CENTROIDS)
+ // *************************************************************************
+
+ private static DataSet<Centroid> getCentroidDataSet(ParameterTool params, ExecutionEnvironment env) {
+ DataSet<Centroid> centroids;
+ if (params.has("centroids")) {
+ centroids = env.readCsvFile(params.get("centroids"))
+ .fieldDelimiter(" ")
+ .pojoType(Centroid.class, "id", "x", "y");
+ } else {
+ System.out.println("Executing K-Means example with default centroid data set.");
+ System.out.println("Use --centroids to specify file input.");
+ centroids = KMeansData.getDefaultCentroidDataSet(env);
+ }
+ return centroids;
+ }
+
+ private static DataSet<Point> getPointDataSet(ParameterTool params, ExecutionEnvironment env) {
+ DataSet<Point> points;
+ if (params.has("points")) {
+ // read points from CSV file
+ points = env.readCsvFile(params.get("points"))
+ .fieldDelimiter(" ")
+ .pojoType(Point.class, "x", "y");
+ } else {
+ System.out.println("Executing K-Means example with default point data set.");
+ System.out.println("Use --points to specify file input.");
+ points = KMeansData.getDefaultPointDataSet(env);
+ }
+ return points;
+ }
+
// *************************************************************************
// DATA TYPES
// *************************************************************************
-
+
/**
* A simple two-dimensional point.
*/
public static class Point implements Serializable {
-
+
public double x, y;
-
+
public Point() {}
public Point(double x, double y) {
this.x = x;
this.y = y;
}
-
+
public Point add(Point other) {
x += other.x;
y += other.y;
return this;
}
-
+
public Point div(long val) {
x /= val;
y /= val;
return this;
}
-
+
public double euclideanDistance(Point other) {
return Math.sqrt((x-other.x)*(x-other.x) + (y-other.y)*(y-other.y));
}
-
+
public void clear() {
x = y = 0.0;
}
-
+
@Override
public String toString() {
return x + " " + y;
}
}
-
+
/**
- * A simple two-dimensional centroid, basically a point with an ID.
+ * A simple two-dimensional centroid, basically a point with an ID.
*/
public static class Centroid extends Point {
-
+
public int id;
-
+
public Centroid() {}
-
+
public Centroid(int id, double x, double y) {
super(x,y);
this.id = id;
}
-
+
public Centroid(int id, Point p) {
super(p.x, p.y);
this.id = id;
}
-
+
@Override
public String toString() {
return id + " " + super.toString();
}
}
-
+
// *************************************************************************
// USER FUNCTIONS
// *************************************************************************
-
- /** Converts a {@code Tuple2<Double,Double>} into a Point. */
- @ForwardedFields("0->x; 1->y")
- public static final class TuplePointConverter implements MapFunction<Tuple2<Double, Double>, Point> {
- @Override
- public Point map(Tuple2<Double, Double> t) throws Exception {
- return new Point(t.f0, t.f1);
- }
- }
-
- /** Converts a {@code Tuple3<Integer, Double,Double>} into a Centroid. */
- @ForwardedFields("0->id; 1->x; 2->y")
- public static final class TupleCentroidConverter implements MapFunction<Tuple3<Integer, Double, Double>, Centroid> {
-
- @Override
- public Centroid map(Tuple3<Integer, Double, Double> t) throws Exception {
- return new Centroid(t.f0, t.f1, t.f2);
- }
- }
-
/** Determines the closest cluster center for a data point. */
@ForwardedFields("*->1")
public static final class SelectNearestCenter extends RichMapFunction<Point, Tuple2<Integer, Point>> {
@@ -228,19 +240,19 @@ public class KMeans {
public void open(Configuration parameters) throws Exception {
this.centroids = getRuntimeContext().getBroadcastVariable("centroids");
}
-
+
@Override
public Tuple2<Integer, Point> map(Point p) throws Exception {
-
+
double minDistance = Double.MAX_VALUE;
int closestCentroidId = -1;
-
+
// check all cluster centers
for (Centroid centroid : centroids) {
// compute distance
double distance = p.euclideanDistance(centroid);
-
- // update nearest cluster if necessary
+
+ // update nearest cluster if necessary
if (distance < minDistance) {
minDistance = distance;
closestCentroidId = centroid.id;
@@ -248,30 +260,30 @@ public class KMeans {
}
// emit a new record with the center id and the data point.
- return new Tuple2<Integer, Point>(closestCentroidId, p);
+ return new Tuple2<>(closestCentroidId, p);
}
}
-
+
/** Appends a count variable to the tuple. */
@ForwardedFields("f0;f1")
public static final class CountAppender implements MapFunction<Tuple2<Integer, Point>, Tuple3<Integer, Point, Long>> {
@Override
public Tuple3<Integer, Point, Long> map(Tuple2<Integer, Point> t) {
- return new Tuple3<Integer, Point, Long>(t.f0, t.f1, 1L);
- }
+ return new Tuple3<>(t.f0, t.f1, 1L);
+ }
}
-
+
/** Sums and counts point coordinates. */
@ForwardedFields("0")
public static final class CentroidAccumulator implements ReduceFunction<Tuple3<Integer, Point, Long>> {
@Override
public Tuple3<Integer, Point, Long> reduce(Tuple3<Integer, Point, Long> val1, Tuple3<Integer, Point, Long> val2) {
- return new Tuple3<Integer, Point, Long>(val1.f0, val1.f1.add(val2.f1), val1.f2 + val2.f2);
+ return new Tuple3<>(val1.f0, val1.f1.add(val2.f1), val1.f2 + val2.f2);
}
}
-
+
/** Computes new centroid from coordinate sum and count of points. */
@ForwardedFields("0->id")
public static final class CentroidAverager implements MapFunction<Tuple3<Integer, Point, Long>, Centroid> {
@@ -281,63 +293,4 @@ public class KMeans {
return new Centroid(value.f0, value.f1.div(value.f2));
}
}
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String pointsPath = null;
- private static String centersPath = null;
- private static String outputPath = null;
- private static int numIterations = 10;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if(programArguments.length > 0) {
- // parse input arguments
- fileOutput = true;
- if(programArguments.length == 4) {
- pointsPath = programArguments[0];
- centersPath = programArguments[1];
- outputPath = programArguments[2];
- numIterations = Integer.parseInt(programArguments[3]);
- } else {
- System.err.println("Usage: KMeans <points path> <centers path> <result path> <num iterations>");
- return false;
- }
- } else {
- System.out.println("Executing K-Means example with default parameters and built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" We provide a data generator to create synthetic input files for this program.");
- System.out.println(" Usage: KMeans <points path> <centers path> <result path> <num iterations>");
- }
- return true;
- }
-
- private static DataSet<Point> getPointDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- // read points from CSV file
- return env.readCsvFile(pointsPath)
- .fieldDelimiter(" ")
- .includeFields(true, true)
- .types(Double.class, Double.class)
- .map(new TuplePointConverter());
- } else {
- return KMeansData.getDefaultPointDataSet(env);
- }
- }
-
- private static DataSet<Centroid> getCentroidDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- return env.readCsvFile(centersPath)
- .fieldDelimiter(" ")
- .includeFields(true, true, true)
- .types(Integer.class, Double.class, Double.class)
- .map(new TupleCentroidConverter());
- } else {
- return KMeansData.getDefaultCentroidDataSet(env);
- }
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
index 8e87892..1d07c2d 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/distcp/DistCp.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.java.LocalEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.operators.FlatMapOperator;
import org.apache.flink.api.java.typeutils.GenericTypeInfo;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.core.fs.FSDataInputStream;
import org.apache.flink.core.fs.FSDataOutputStream;
@@ -62,18 +63,32 @@ public class DistCp {
public static final String FILES_COPIED_CNT_NAME = "FILES_COPIED";
public static void main(String[] args) throws Exception {
- if (args.length != 3) {
- printHelp();
+
+ // set up the execution environment
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ ParameterTool params = ParameterTool.fromArgs(args);
+ if (!params.has("input") || !params.has("output")) {
+ System.err.println("Usage: --input <path> --output <path> [--parallelism <n>]");
return;
}
- final Path sourcePath = new Path(args[0]);
- final Path targetPath = new Path(args[1]);
- int parallelism = Integer.valueOf(args[2], 10);
+ final Path sourcePath = new Path(params.get("input"));
+ final Path targetPath = new Path(params.get("output"));
+ if (!isLocal(env) && !(isOnDistributedFS(sourcePath) && isOnDistributedFS(targetPath))) {
+ System.out.println("In a distributed mode only HDFS input/output paths are supported");
+ return;
+ }
+
+ final int parallelism = params.getInt("parallelism", 10);
+ if (parallelism <= 0) {
+ System.err.println("Parallelism should be greater than 0");
+ return;
+ }
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
- // set up the execution environment
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- checkInputParams(env, sourcePath, targetPath, parallelism);
env.setParallelism(parallelism);
long startTime = System.currentTimeMillis();
@@ -142,20 +157,12 @@ public class DistCp {
// HELPER METHODS
// -----------------------------------------------------------------------------------------
- private static void checkInputParams(ExecutionEnvironment env, Path sourcePath, Path targetPath, int parallelism) throws IOException {
- if (parallelism <= 0) {
- throw new IllegalArgumentException("Parallelism should be greater than 0");
- }
-
- boolean isLocal = env instanceof LocalEnvironment;
- if (!isLocal &&
- !(sourcePath.getFileSystem().isDistributedFS() && targetPath.getFileSystem().isDistributedFS())) {
- throw new IllegalArgumentException("In a distributed mode only HDFS input/output paths are supported");
- }
+ private static boolean isLocal(final ExecutionEnvironment env) {
+ return env instanceof LocalEnvironment;
}
- private static void printHelp() {
- System.err.println("Usage: <input_path> <output_path> <level_of_parallelism>");
+ private static boolean isOnDistributedFS(final Path path) throws IOException {
+ return path.getFileSystem().isDistributedFS();
}
private static List<FileCopyTask> getCopyTasks(Path sourcePath) throws IOException {
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
index ce36504..a86cdfd 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/ConnectedComponents.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFieldsFir
import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFieldsSecond;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
import org.apache.flink.util.Collector;
import org.apache.flink.api.java.DataSet;
@@ -62,7 +63,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
* </ul>
*
* <p>
- * Usage: <code>ConnectedComponents <vertices path> <edges path> <result path> <max number of iterations></code><br>
+ * Usage: <code>ConnectedComponents --vertices <path> --edges <path> --output <path> --iterations <n></code><br>
* If no parameters are provided, the program is run with default data from {@link org.apache.flink.examples.java.graph.util.ConnectedComponentsData} and 10 iterations.
*
* <p>
@@ -80,20 +81,26 @@ public class ConnectedComponents implements ProgramDescription {
// *************************************************************************
public static void main(String... args) throws Exception {
-
- if(!parseParameters(args)) {
- return;
- }
-
+
+ // Checking input parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: ConnectedComponents --vertices <path> --edges <path> --output <path> --iterations <n>");
+
// set up execution environment
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ final int maxIterations = params.getInt("iterations", 10);
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
// read vertex and edge data
- DataSet<Long> vertices = getVertexDataSet(env);
- DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env).flatMap(new UndirectEdge());
+ DataSet<Long> vertices = getVertexDataSet(env, params);
+ DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env, params).flatMap(new UndirectEdge());
// assign the initial components (equal to the vertex id)
- DataSet<Tuple2<Long, Long>> verticesWithInitialId = vertices.map(new DuplicateValue<Long>());
+ DataSet<Tuple2<Long, Long>> verticesWithInitialId =
+ vertices.map(new DuplicateValue<Long>());
// open a delta iteration
DeltaIteration<Tuple2<Long, Long>, Tuple2<Long, Long>> iteration =
@@ -109,11 +116,12 @@ public class ConnectedComponents implements ProgramDescription {
DataSet<Tuple2<Long, Long>> result = iteration.closeWith(changes, changes);
// emit result
- if (fileOutput) {
- result.writeAsCsv(outputPath, "\n", " ");
+ if (params.has("output")) {
+ result.writeAsCsv(params.get("output"), "\n", " ");
// execute program
env.execute("Connected Components Example");
} else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
result.print();
}
}
@@ -181,60 +189,34 @@ public class ConnectedComponents implements ProgramDescription {
@Override
public String getDescription() {
- return "Parameters: <vertices-path> <edges-path> <result-path> <max-number-of-iterations>";
+ return "Parameters: --vertices <path> --edges <path> --output <path> --iterations <n>";
}
// *************************************************************************
// UTIL METHODS
// *************************************************************************
-
- private static boolean fileOutput = false;
- private static String verticesPath = null;
- private static String edgesPath = null;
- private static String outputPath = null;
- private static int maxIterations = 10;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if(programArguments.length > 0) {
- // parse input arguments
- fileOutput = true;
- if(programArguments.length == 4) {
- verticesPath = programArguments[0];
- edgesPath = programArguments[1];
- outputPath = programArguments[2];
- maxIterations = Integer.parseInt(programArguments[3]);
- } else {
- System.err.println("Usage: ConnectedComponents <vertices path> <edges path> <result path> <max number of iterations>");
- return false;
- }
- } else {
- System.out.println("Executing Connected Components example with default parameters and built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" Usage: ConnectedComponents <vertices path> <edges path> <result path> <max number of iterations>");
- }
- return true;
- }
-
- private static DataSet<Long> getVertexDataSet(ExecutionEnvironment env) {
-
- if(fileOutput) {
- return env.readCsvFile(verticesPath).types(Long.class)
- .map(
- new MapFunction<Tuple1<Long>, Long>() {
- public Long map(Tuple1<Long> value) { return value.f0; }
- });
+
+ private static DataSet<Long> getVertexDataSet(ExecutionEnvironment env, ParameterTool params) {
+ if (params.has("vertices")) {
+ return env.readCsvFile(params.get("vertices")).types(Long.class).map(
+ new MapFunction<Tuple1<Long>, Long>() {
+ public Long map(Tuple1<Long> value) {
+ return value.f0;
+ }
+ });
} else {
+ System.out.println("Executing Connected Components example with default vertices data set.");
+ System.out.println("Use --vertices to specify file input.");
return ConnectedComponentsData.getDefaultVertexDataSet(env);
}
}
-
- private static DataSet<Tuple2<Long, Long>> getEdgeDataSet(ExecutionEnvironment env) {
-
- if(fileOutput) {
- return env.readCsvFile(edgesPath).fieldDelimiter(" ").types(Long.class, Long.class);
+
+ private static DataSet<Tuple2<Long, Long>> getEdgeDataSet(ExecutionEnvironment env, ParameterTool params) {
+ if (params.has("edges")) {
+ return env.readCsvFile(params.get("edges")).fieldDelimiter(" ").types(Long.class, Long.class);
} else {
+ System.out.println("Executing Connected Components example with default edges data set.");
+ System.out.println("Use --edges to specify file input.");
return ConnectedComponentsData.getDefaultEdgeDataSet(env);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/EnumTriangles.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/EnumTriangles.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/EnumTriangles.java
index b789008..33570fc 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/EnumTriangles.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/EnumTriangles.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.examples.java.graph.util.EnumTrianglesData;
import org.apache.flink.examples.java.graph.util.EnumTrianglesDataTypes.Edge;
import org.apache.flink.examples.java.graph.util.EnumTrianglesDataTypes.Triad;
@@ -59,7 +60,7 @@ import java.util.List;
* (2)-(12)
* </pre>
*
- * Usage: <code>EnumTriangleBasic <edge path> <result path></code><br>
+ * Usage: <code>EnumTriangleBasic --edges <path> --output <path></code><br>
* If no parameters are provided, the program is run with default data from {@link EnumTrianglesData}.
*
* <p>
@@ -73,26 +74,36 @@ import java.util.List;
@SuppressWarnings("serial")
public class EnumTriangles {
- static boolean fileOutput = false;
- static String edgePath = null;
- static String outputPath = null;
-
// *************************************************************************
// PROGRAM
// *************************************************************************
public static void main(String[] args) throws Exception {
-
- if(!parseParameters(args)) {
- return;
- }
-
+
+ // Checking input parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: EnumTriangleBasic --edges <path> --output <path>");
+
// set up execution environment
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
// read input data
- DataSet<Edge> edges = getEdgeDataSet(env);
-
+ DataSet<Edge> edges;
+ if (params.has("edges")) {
+ edges = env.readCsvFile(params.get("edges"))
+ .fieldDelimiter(" ")
+ .includeFields(true, true)
+ .types(Integer.class, Integer.class)
+ .map(new TupleEdgeConverter());
+ } else {
+ System.out.println("Executing EnumTriangles example with default edges data set.");
+ System.out.println("Use --edges to specify file input.");
+ edges = EnumTrianglesData.getDefaultEdgeDataSet(env);
+ }
+
// project edges by vertex id
DataSet<Edge> edgesById = edges
.map(new EdgeByIdProjector());
@@ -104,11 +115,12 @@ public class EnumTriangles {
.join(edgesById).where(Triad.V2, Triad.V3).equalTo(Edge.V1, Edge.V2).with(new TriadFilter());
// emit result
- if (fileOutput) {
- triangles.writeAsCsv(outputPath, "\n", ",");
+ if (params.has("output")) {
+ triangles.writeAsCsv(params.get("output"), "\n", ",");
// execute program
env.execute("Basic Triangle Enumeration Example");
} else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
triangles.print();
}
}
@@ -191,41 +203,4 @@ public class EnumTriangles {
}
}
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean parseParameters(String[] args) {
-
- if(args.length > 0) {
- // parse input arguments
- fileOutput = true;
- if(args.length == 2) {
- edgePath = args[0];
- outputPath = args[1];
- } else {
- System.err.println("Usage: EnumTriangleBasic <edge path> <result path>");
- return false;
- }
- } else {
- System.out.println("Executing Enum Triangles Basic example with built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" Usage: EnumTriangleBasic <edge path> <result path>");
- }
- return true;
- }
-
- private static DataSet<Edge> getEdgeDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- return env.readCsvFile(edgePath)
- .fieldDelimiter(" ")
- .includeFields(true, true)
- .types(Integer.class, Integer.class)
- .map(new TupleEdgeConverter());
- } else {
- return EnumTrianglesData.getDefaultEdgeDataSet(env);
- }
- }
-
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java
index a2ea511..f8a5c83 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/PageRank.java
@@ -28,6 +28,7 @@ import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
import org.apache.flink.api.java.operators.IterativeDataSet;
import org.apache.flink.api.java.tuple.Tuple1;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.examples.java.graph.util.PageRankData;
import org.apache.flink.util.Collector;
@@ -58,7 +59,7 @@ import static org.apache.flink.api.java.aggregation.Aggregations.SUM;
* </ul>
*
* <p>
- * Usage: <code>PageRankBasic <pages path> <links path> <output path> <num pages> <num iterations></code><br>
+ * Usage: <code>PageRankBasic --pages <path> --links <path> --output <path> --numPages <n> --iterations <n></code><br>
* If no parameters are provided, the program is run with default data from {@link org.apache.flink.examples.java.graph.util.PageRankData} and 10 iterations.
*
* <p>
@@ -80,17 +81,22 @@ public class PageRank {
// *************************************************************************
public static void main(String[] args) throws Exception {
-
- if(!parseParameters(args)) {
- return;
- }
+
+ ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: PageRankBasic --pages <path> --links <path> --output <path> --numPages <n> --iterations <n>");
+
+ final int numPages = params.getInt("numPages");
+ final int maxIterations = params.getInt("iterations", 10);
// set up execution environment
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ // make the parameters available to the web ui
+ env.getConfig().setGlobalJobParameters(params);
// get input data
- DataSet<Long> pagesInput = getPagesDataSet(env);
- DataSet<Tuple2<Long, Long>> linksInput = getLinksDataSet(env);
+ DataSet<Long> pagesInput = getPagesDataSet(env, params);
+ DataSet<Tuple2<Long, Long>> linksInput = getLinksDataSet(env, params);
// assign initial rank to pages
DataSet<Tuple2<Long, Double>> pagesWithRanks = pagesInput.
@@ -118,11 +124,12 @@ public class PageRank {
.filter(new EpsilonFilter()));
// emit result
- if(fileOutput) {
- finalPageRanks.writeAsCsv(outputPath, "\n", " ");
+ if (params.has("output")) {
+ finalPageRanks.writeAsCsv(params.get("output"), "\n", " ");
// execute program
env.execute("Basic Page Rank Example");
} else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
finalPageRanks.print();
}
@@ -179,12 +186,12 @@ public class PageRank {
@Override
public void flatMap(Tuple2<Tuple2<Long, Double>, Tuple2<Long, Long[]>> value, Collector<Tuple2<Long, Double>> out){
- Long[] neigbors = value.f1.f1;
+ Long[] neighbors = value.f1.f1;
double rank = value.f0.f1;
- double rankToDistribute = rank / ((double) neigbors.length);
+ double rankToDistribute = rank / ((double) neighbors.length);
- for (int i = 0; i < neigbors.length; i++) {
- out.collect(new Tuple2<Long, Double>(neigbors[i], rankToDistribute));
+ for (int i = 0; i < neighbors.length; i++) {
+ out.collect(new Tuple2<Long, Double>(neighbors[i], rankToDistribute));
}
}
}
@@ -225,61 +232,34 @@ public class PageRank {
// UTIL METHODS
// *************************************************************************
- private static boolean fileOutput = false;
- private static String pagesInputPath = null;
- private static String linksInputPath = null;
- private static String outputPath = null;
- private static long numPages = 0;
- private static int maxIterations = 10;
-
- private static boolean parseParameters(String[] args) {
-
- if(args.length > 0) {
- if(args.length == 5) {
- fileOutput = true;
- pagesInputPath = args[0];
- linksInputPath = args[1];
- outputPath = args[2];
- numPages = Integer.parseInt(args[3]);
- maxIterations = Integer.parseInt(args[4]);
- } else {
- System.err.println("Usage: PageRankBasic <pages path> <links path> <output path> <num pages> <num iterations>");
- return false;
- }
- } else {
- System.out.println("Executing PageRank Basic example with default parameters and built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" Usage: PageRankBasic <pages path> <links path> <output path> <num pages> <num iterations>");
-
- numPages = PageRankData.getNumberOfPages();
- }
- return true;
- }
-
- private static DataSet<Long> getPagesDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- return env
- .readCsvFile(pagesInputPath)
- .fieldDelimiter(" ")
- .lineDelimiter("\n")
- .types(Long.class)
- .map(new MapFunction<Tuple1<Long>, Long>() {
- @Override
- public Long map(Tuple1<Long> v) { return v.f0; }
- });
+ private static DataSet<Long> getPagesDataSet(ExecutionEnvironment env, ParameterTool params) {
+ if (params.has("pages")) {
+ return env.readCsvFile(params.get("pages"))
+ .fieldDelimiter(" ")
+ .lineDelimiter("\n")
+ .types(Long.class)
+ .map(new MapFunction<Tuple1<Long>, Long>() {
+ @Override
+ public Long map(Tuple1<Long> v) {
+ return v.f0;
+ }
+ });
} else {
+ System.out.println("Executing PageRank example with default pages data set.");
+ System.out.println("Use --pages to specify file input.");
return PageRankData.getDefaultPagesDataSet(env);
}
}
-
- private static DataSet<Tuple2<Long, Long>> getLinksDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- return env.readCsvFile(linksInputPath)
- .fieldDelimiter(" ")
- .lineDelimiter("\n")
- .types(Long.class, Long.class);
+
+ private static DataSet<Tuple2<Long, Long>> getLinksDataSet(ExecutionEnvironment env, ParameterTool params) {
+ if (params.has("links")) {
+ return env.readCsvFile(params.get("links"))
+ .fieldDelimiter(" ")
+ .lineDelimiter("\n")
+ .types(Long.class, Long.class);
} else {
+ System.out.println("Executing PageRank example with default links data set.");
+ System.out.println("Use --links to specify file input.");
return PageRankData.getDefaultEdgeDataSet(env);
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/TransitiveClosureNaive.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/TransitiveClosureNaive.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/TransitiveClosureNaive.java
index 2ed453e..b4cf036 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/TransitiveClosureNaive.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/graph/TransitiveClosureNaive.java
@@ -26,6 +26,7 @@ import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.IterativeDataSet;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.examples.java.graph.util.ConnectedComponentsData;
import org.apache.flink.util.Collector;
@@ -35,17 +36,28 @@ import java.util.Set;
@SuppressWarnings("serial")
public class TransitiveClosureNaive implements ProgramDescription {
+ public static void main (String... args) throws Exception {
- public static void main (String... args) throws Exception{
-
- if (!parseParameters(args)) {
- return;
- }
+ // Checking input parameters
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: TransitiveClosure --edges <path> --output <path> --iterations <n>");
// set up execution environment
ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
- DataSet<Tuple2<Long, Long>> edges = getEdgeDataSet(env);
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
+
+ final int maxIterations = params.getInt("iterations", 10);
+
+ DataSet<Tuple2<Long, Long>> edges;
+ if (params.has("edges")) {
+ edges = env.readCsvFile(params.get("edges")).fieldDelimiter(" ").types(Long.class, Long.class);
+ } else {
+ System.out.println("Executing TransitiveClosureNaive example with default edges data set.");
+ System.out.println("Use --edges to specify file input.");
+ edges = ConnectedComponentsData.getDefaultEdgeDataSet(env);
+ }
IterativeDataSet<Tuple2<Long,Long>> paths = edges.iterate(maxIterations);
@@ -95,60 +107,20 @@ public class TransitiveClosureNaive implements ProgramDescription {
// emit result
- if (fileOutput) {
- transitiveClosure.writeAsCsv(outputPath, "\n", " ");
+ if (params.has("output")) {
+ transitiveClosure.writeAsCsv(params.get("output"), "\n", " ");
// execute program explicitly, because file sinks are lazy
env.execute("Transitive Closure Example");
} else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
transitiveClosure.print();
}
}
@Override
public String getDescription() {
- return "Parameters: <edges-path> <result-path> <max-number-of-iterations>";
- }
-
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String edgesPath = null;
- private static String outputPath = null;
- private static int maxIterations = 10;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if (programArguments.length > 0) {
- // parse input arguments
- fileOutput = true;
- if (programArguments.length == 3) {
- edgesPath = programArguments[0];
- outputPath = programArguments[1];
- maxIterations = Integer.parseInt(programArguments[2]);
- } else {
- System.err.println("Usage: TransitiveClosure <edges path> <result path> <max number of iterations>");
- return false;
- }
- } else {
- System.out.println("Executing TransitiveClosure example with default parameters and built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" Usage: TransitiveClosure <edges path> <result path> <max number of iterations>");
- }
- return true;
- }
-
-
- private static DataSet<Tuple2<Long, Long>> getEdgeDataSet(ExecutionEnvironment env) {
-
- if(fileOutput) {
- return env.readCsvFile(edgesPath).fieldDelimiter(" ").types(Long.class, Long.class);
- } else {
- return ConnectedComponentsData.getDefaultEdgeDataSet(env);
- }
+ return "Parameters: --edges <path> --output <path> --iterations <n>";
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
index 77470e7..d225215 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/LinearRegression.java
@@ -24,8 +24,8 @@ import java.util.Collection;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
-import org.apache.flink.api.java.functions.FunctionAnnotation.ForwardedFields;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.examples.java.ml.util.LinearRegressionData;
import org.apache.flink.api.java.DataSet;
@@ -70,24 +70,38 @@ public class LinearRegression {
// PROGRAM
// *************************************************************************
- public static void main(String[] args) throws Exception{
+ public static void main(String[] args) throws Exception {
- if(!parseParameters(args)) {
- return;
- }
+ final ParameterTool params = ParameterTool.fromArgs(args);
+ System.out.println("Usage: LinearRegression --input <path> --output <path> --iterations <n>");
// set up execution environment
-
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ final int iterations = params.getInt("iterations", 10);
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
+
// get input x data from elements
- DataSet<Data> data = getDataSet(env);
+ DataSet<Data> data;
+ if (params.has("input")) {
+ // read data from CSV file
+ data = env.readCsvFile(params.get("input"))
+ .fieldDelimiter(" ")
+ .includeFields(true, true)
+ .pojoType(Data.class);
+ } else {
+ System.out.println("Executing LinearRegression example with default input data set.");
+ System.out.println("Use --input to specify file input.");
+ data = LinearRegressionData.getDefaultDataDataSet(env);
+ }
// get the parameters from elements
- DataSet<Params> parameters = getParamsDataSet(env);
+ DataSet<Params> parameters = LinearRegressionData.getDefaultParamsDataSet(env);
// set number of bulk iterations for SGD linear Regression
- IterativeDataSet<Params> loop = parameters.iterate(numIterations);
+ IterativeDataSet<Params> loop = parameters.iterate(iterations);
DataSet<Params> new_parameters = data
// compute a single step using every sample
@@ -101,11 +115,12 @@ public class LinearRegression {
DataSet<Params> result = loop.closeWith(new_parameters);
// emit result
- if(fileOutput) {
- result.writeAsText(outputPath);
+ if(params.has("output")) {
+ result.writeAsText(params.get("output"));
// execute program
env.execute("Linear Regression example");
} else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
result.print();
}
}
@@ -181,26 +196,6 @@ public class LinearRegression {
// USER FUNCTIONS
// *************************************************************************
- /** Converts a {@code Tuple2<Double,Double>} into a Data. */
- @ForwardedFields("0->x; 1->y")
- public static final class TupleDataConverter implements MapFunction<Tuple2<Double, Double>, Data> {
-
- @Override
- public Data map(Tuple2<Double, Double> t) throws Exception {
- return new Data(t.f0, t.f1);
- }
- }
-
- /** Converts a {@code Tuple2<Double,Double>} into a Params. */
- @ForwardedFields("0->theta0; 1->theta1")
- public static final class TupleParamsConverter implements MapFunction<Tuple2<Double, Double>,Params> {
-
- @Override
- public Params map(Tuple2<Double, Double> t)throws Exception {
- return new Params(t.f0,t.f1);
- }
- }
-
/**
* Compute a single BGD type update for every parameters.
*/
@@ -261,56 +256,6 @@ public class LinearRegression {
}
}
- // *************************************************************************
- // UTIL METHODS
- // *************************************************************************
-
- private static boolean fileOutput = false;
- private static String dataPath = null;
- private static String outputPath = null;
- private static int numIterations = 10;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if(programArguments.length > 0) {
- // parse input arguments
- fileOutput = true;
- if(programArguments.length == 3) {
- dataPath = programArguments[0];
- outputPath = programArguments[1];
- numIterations = Integer.parseInt(programArguments[2]);
- } else {
- System.err.println("Usage: LinearRegression <data path> <result path> <num iterations>");
- return false;
- }
- } else {
- System.out.println("Executing Linear Regression example with default parameters and built-in default data.");
- System.out.println(" Provide parameters to read input data from files.");
- System.out.println(" See the documentation for the correct format of input files.");
- System.out.println(" We provide a data generator to create synthetic input files for this program.");
- System.out.println(" Usage: LinearRegression <data path> <result path> <num iterations>");
- }
- return true;
- }
-
- private static DataSet<Data> getDataSet(ExecutionEnvironment env) {
- if(fileOutput) {
- // read data from CSV file
- return env.readCsvFile(dataPath)
- .fieldDelimiter(" ")
- .includeFields(true, true)
- .types(Double.class, Double.class)
- .map(new TupleDataConverter());
- } else {
- return LinearRegressionData.getDefaultDataDataSet(env);
- }
- }
-
- private static DataSet<Params> getParamsDataSet(ExecutionEnvironment env) {
-
- return LinearRegressionData.getDefaultParamsDataSet(env);
-
- }
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
index 73b4b55..838e320 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/ml/util/LinearRegressionData.java
@@ -51,8 +51,7 @@ public class LinearRegressionData {
new Object[] { 0.28, 0.57 }, new Object[] { 1.65, 3.30 },
new Object[] { -0.55, -1.08 } };
- public static DataSet<Params> getDefaultParamsDataSet(
- ExecutionEnvironment env) {
+ public static DataSet<Params> getDefaultParamsDataSet(ExecutionEnvironment env) {
List<Params> paramsList = new LinkedList<>();
for (Object[] params : PARAMS) {
paramsList.add(new Params((Double) params[0], (Double) params[1]));
@@ -61,7 +60,6 @@ public class LinearRegressionData {
}
public static DataSet<Data> getDefaultDataDataSet(ExecutionEnvironment env) {
-
List<Data> dataList = new LinkedList<>();
for (Object[] data : DATA) {
dataList.add(new Data((Double) data[0], (Double) data[1]));
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/EmptyFieldsCountAccumulator.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/EmptyFieldsCountAccumulator.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/EmptyFieldsCountAccumulator.java
index 1578fd0..b6eba8f 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/EmptyFieldsCountAccumulator.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/EmptyFieldsCountAccumulator.java
@@ -29,8 +29,8 @@ import org.apache.flink.api.common.accumulators.Accumulator;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.common.functions.RichFilterFunction;
-import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple3;
+import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
/**
@@ -39,15 +39,16 @@ import org.apache.flink.configuration.Configuration;
* most contain whitespace characters like space and tab.
* <p>
* The input file is a plain text CSV file with the semicolon as field separator and double quotes as field delimiters
- * and three columns. See {@link #getDataSet(ExecutionEnvironment)} for configuration.
+ * and three columns. See {@link #getDataSet(ExecutionEnvironment, ParameterTool)} for configuration.
* <p>
- * Usage: <code>FilterAndCountIncompleteLines [<input file path> [<result path>]]</code> <br>
+ * Usage: <code>EmptyFieldsCountAccumulator --input <path> --output <path></code> <br>
* <p>
* This example shows how to use:
* <ul>
* <li>custom accumulators
* <li>tuple data types
* <li>inline-defined functions
+ * <li>naming large tuple types
* </ul>
*/
@SuppressWarnings("serial")
@@ -61,28 +62,31 @@ public class EmptyFieldsCountAccumulator {
public static void main(final String[] args) throws Exception {
- if (!parseParameters(args)) {
- return;
- }
+ final ParameterTool params = ParameterTool.fromArgs(args);
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ System.out.println("Usage: EmptyFieldsCountAccumulator --input <path> --output <path>");
+
+ // make parameters available in the web interface
+ env.getConfig().setGlobalJobParameters(params);
// get the data set
- final DataSet<Tuple> file = getDataSet(env);
+ final DataSet<StringTriple> file = getDataSet(env, params);
// filter lines with empty fields
- final DataSet<Tuple> filteredLines = file.filter(new EmptyFieldFilter());
+ final DataSet<StringTriple> filteredLines = file.filter(new EmptyFieldFilter());
// Here, we could do further processing with the filtered lines...
JobExecutionResult result;
// output the filtered lines
- if (outputPath == null) {
- filteredLines.print();
- result = env.getLastJobExecutionResult();
- } else {
- filteredLines.writeAsCsv(outputPath);
+ if (params.has("output")) {
+ filteredLines.writeAsCsv(params.get("output"));
// execute program
result = env.execute("Accumulator example");
+ } else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
+ filteredLines.print();
+ result = env.getLastJobExecutionResult();
}
// get the accumulator result via its registration key
@@ -94,51 +98,26 @@ public class EmptyFieldsCountAccumulator {
// UTIL METHODS
// *************************************************************************
- private static String filePath;
- private static String outputPath;
-
- private static boolean parseParameters(final String[] programArguments) {
-
- if (programArguments.length >= 3) {
- System.err.println("Usage: FilterAndCountIncompleteLines [<input file path> [<result path>]]");
- return false;
- }
-
- if (programArguments.length >= 1) {
- filePath = programArguments[0];
- if (programArguments.length == 2) {
- outputPath = programArguments[1];
- }
- }
-
- return true;
- }
-
@SuppressWarnings("unchecked")
- private static DataSet<Tuple> getDataSet(final ExecutionEnvironment env) {
-
- DataSet<? extends Tuple> source;
- if (filePath == null) {
- source = env.fromCollection(getExampleInputTuples());
-
+ private static DataSet<StringTriple> getDataSet(ExecutionEnvironment env, ParameterTool params) {
+ if (params.has("input")) {
+ return env.readCsvFile(params.get("input"))
+ .fieldDelimiter(";")
+ .pojoType(StringTriple.class);
} else {
- source = env
- .readCsvFile(filePath)
- .fieldDelimiter(";")
- .types(String.class, String.class, String.class);
-
+ System.out.println("Executing EmptyFieldsCountAccumulator example with default input data set.");
+ System.out.println("Use --input to specify file input.");
+ return env.fromCollection(getExampleInputTuples());
}
-
- return (DataSet<Tuple>) source;
}
- private static Collection<Tuple3<String, String, String>> getExampleInputTuples() {
- Collection<Tuple3<String, String, String>> inputTuples = new ArrayList<Tuple3<String, String, String>>();
- inputTuples.add(new Tuple3<String, String, String>("John", "Doe", "Foo Str."));
- inputTuples.add(new Tuple3<String, String, String>("Joe", "Johnson", ""));
- inputTuples.add(new Tuple3<String, String, String>(null, "Kate Morn", "Bar Blvd."));
- inputTuples.add(new Tuple3<String, String, String>("Tim", "Rinny", ""));
- inputTuples.add(new Tuple3<String, String, String>("Alicia", "Jackson", " "));
+ private static Collection<StringTriple> getExampleInputTuples() {
+ Collection<StringTriple> inputTuples = new ArrayList<StringTriple>();
+ inputTuples.add(new StringTriple("John", "Doe", "Foo Str."));
+ inputTuples.add(new StringTriple("Joe", "Johnson", ""));
+ inputTuples.add(new StringTriple(null, "Kate Morn", "Bar Blvd."));
+ inputTuples.add(new StringTriple("Tim", "Rinny", ""));
+ inputTuples.add(new StringTriple("Alicia", "Jackson", " "));
return inputTuples;
}
@@ -147,7 +126,7 @@ public class EmptyFieldsCountAccumulator {
* In doing so, it also counts the number of empty fields per attribute with an accumulator (registered under
* {@link EmptyFieldsCountAccumulator#EMPTY_FIELD_ACCUMULATOR}).
*/
- public static final class EmptyFieldFilter extends RichFilterFunction<Tuple> {
+ public static final class EmptyFieldFilter extends RichFilterFunction<StringTriple> {
// create a new accumulator in each filter function instance
// accumulators can be merged later on
@@ -163,7 +142,7 @@ public class EmptyFieldsCountAccumulator {
}
@Override
- public boolean filter(final Tuple t) {
+ public boolean filter(final StringTriple t) {
boolean containsEmptyFields = false;
// iterate over the tuple fields looking for empty ones
@@ -252,4 +231,20 @@ public class EmptyFieldsCountAccumulator {
return StringUtils.join(resultVector, ',');
}
}
+
+ /**
+ * It is recommended to use POJOs (Plain old Java objects) instead of TupleX for
+ * data types with many fields. Also, POJOs can be used to give large Tuple-types a name.
+ * <a href="https://ci.apache.org/projects/flink/flink-docs-master/apis/best_practices.html#naming-large-tuplex-types">Source (docs)</a>
+ */
+ public static class StringTriple extends Tuple3<String, String, String> {
+
+ public StringTriple() {}
+
+ public StringTriple(String f0, String f1, String f2) {
+ super(f0, f1, f2);
+ }
+
+ }
+
}
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java
index 495ff26..816be1e 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery10.java
@@ -29,6 +29,7 @@ import org.apache.flink.api.java.tuple.Tuple6;
import org.apache.flink.api.java.DataSet;
import org.apache.flink.api.java.ExecutionEnvironment;
+import org.apache.flink.api.java.utils.ParameterTool;
/**
* This program implements a modified version of the TPC-H query 10.
@@ -76,7 +77,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
* as generated by the TPC-H data generator which is available at <a href="http://www.tpc.org/tpch/">http://www.tpc.org/tpch/</a>.
*
* <p>
- * Usage: <code>TPCHQuery10 <customer-csv path> <orders-csv path> <lineitem-csv path> <nation-csv path> <result path></code><br>
+ * Usage: <code>TPCHQuery10 --customer <path> --orders <path> --lineitem<path> --nation <path> --output <path></code><br>
*
* <p>
* This example shows how to use:
@@ -84,7 +85,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
* <li> tuple data types
* <li> inline-defined functions
* <li> projection and join projection
- * <li> build-in aggregation functions
+ * <li> built-in aggregation functions
* </ul>
*/
@SuppressWarnings("serial")
@@ -95,21 +96,31 @@ public class TPCHQuery10 {
// *************************************************************************
public static void main(String[] args) throws Exception {
-
- if(!parseParameters(args)) {
+
+ final ParameterTool params = ParameterTool.fromArgs(args);
+
+ final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+
+ if (!params.has("customer") && !params.has("orders") && !params.has("lineitem") && !params.has("nation")) {
+ System.err.println(" This program expects data from the TPC-H benchmark as input data.");
+ System.err.println(" Due to legal restrictions, we can not ship generated data.");
+ System.err.println(" You can find the TPC-H data generator at http://www.tpc.org/tpch/.");
+ System.err.println(" Usage: TPCHQuery10 --customer <path> --orders <path> --lineitem <path> --nation <path> [--output <path>]");
return;
}
-
- final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
// get customer data set: (custkey, name, address, nationkey, acctbal)
- DataSet<Tuple5<Integer, String, String, Integer, Double>> customers = getCustomerDataSet(env);
+ DataSet<Tuple5<Integer, String, String, Integer, Double>> customers =
+ getCustomerDataSet(env, params.get("customer"));
// get orders data set: (orderkey, custkey, orderdate)
- DataSet<Tuple3<Integer, Integer, String>> orders = getOrdersDataSet(env);
+ DataSet<Tuple3<Integer, Integer, String>> orders =
+ getOrdersDataSet(env, params.get("orders"));
// get lineitem data set: (orderkey, extendedprice, discount, returnflag)
- DataSet<Tuple4<Integer, Double, Double, String>> lineitems = getLineitemDataSet(env);
+ DataSet<Tuple4<Integer, Double, Double, String>> lineitems =
+ getLineitemDataSet(env, params.get("lineitem"));
// get nation data set: (nationkey, name)
- DataSet<Tuple2<Integer, String>> nations = getNationsDataSet(env);
+ DataSet<Tuple2<Integer, String>> nations =
+ getNationsDataSet(env, params.get("nation"));
// orders filtered by year: (orderkey, custkey)
DataSet<Tuple2<Integer, Integer>> ordersFilteredByYear =
@@ -163,10 +174,14 @@ public class TPCHQuery10 {
.projectFirst(0,1,2,3,4).projectSecond(1);
// emit result
- result.writeAsCsv(outputPath, "\n", "|");
-
- // execute program
- env.execute("TPCH Query 10 Example");
+ if (params.has("output")) {
+ result.writeAsCsv(params.get("output"), "\n", "|");
+ // execute program
+ env.execute("TPCH Query 10 Example");
+ } else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
+ result.print();
+ }
}
@@ -174,57 +189,28 @@ public class TPCHQuery10 {
// UTIL METHODS
// *************************************************************************
- private static String customerPath;
- private static String ordersPath;
- private static String lineitemPath;
- private static String nationPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if(programArguments.length > 0) {
- if(programArguments.length == 5) {
- customerPath = programArguments[0];
- ordersPath = programArguments[1];
- lineitemPath = programArguments[2];
- nationPath = programArguments[3];
- outputPath = programArguments[4];
- } else {
- System.err.println("Usage: TPCHQuery10 <customer-csv path> <orders-csv path> <lineitem-csv path> <nation-csv path> <result path>");
- return false;
- }
- } else {
- System.err.println("This program expects data from the TPC-H benchmark as input data.\n" +
- " Due to legal restrictions, we can not ship generated data.\n" +
- " You can find the TPC-H data generator at http://www.tpc.org/tpch/.\n" +
- " Usage: TPCHQuery10 <customer-csv path> <orders-csv path> <lineitem-csv path> <nation-csv path> <result path>");
- return false;
- }
- return true;
- }
-
- private static DataSet<Tuple5<Integer, String, String, Integer, Double>> getCustomerDataSet(ExecutionEnvironment env) {
+ private static DataSet<Tuple5<Integer, String, String, Integer, Double>> getCustomerDataSet(ExecutionEnvironment env, String customerPath) {
return env.readCsvFile(customerPath)
.fieldDelimiter("|")
.includeFields("11110100")
.types(Integer.class, String.class, String.class, Integer.class, Double.class);
}
- private static DataSet<Tuple3<Integer, Integer, String>> getOrdersDataSet(ExecutionEnvironment env) {
+ private static DataSet<Tuple3<Integer, Integer, String>> getOrdersDataSet(ExecutionEnvironment env, String ordersPath) {
return env.readCsvFile(ordersPath)
.fieldDelimiter("|")
.includeFields("110010000")
.types(Integer.class, Integer.class, String.class);
}
- private static DataSet<Tuple4<Integer, Double, Double, String>> getLineitemDataSet(ExecutionEnvironment env) {
+ private static DataSet<Tuple4<Integer, Double, Double, String>> getLineitemDataSet(ExecutionEnvironment env, String lineitemPath) {
return env.readCsvFile(lineitemPath)
.fieldDelimiter("|")
.includeFields("1000011010000000")
.types(Integer.class, Double.class, Double.class, String.class);
}
- private static DataSet<Tuple2<Integer, String>> getNationsDataSet(ExecutionEnvironment env) {
+ private static DataSet<Tuple2<Integer, String>> getNationsDataSet(ExecutionEnvironment env, String nationPath) {
return env.readCsvFile(nationPath)
.fieldDelimiter("|")
.includeFields("1100")
http://git-wip-us.apache.org/repos/asf/flink/blob/0629e256/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery3.java
----------------------------------------------------------------------
diff --git a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery3.java b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery3.java
index fd11c4d..e0bd536 100644
--- a/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery3.java
+++ b/flink-examples/flink-examples-batch/src/main/java/org/apache/flink/examples/java/relational/TPCHQuery3.java
@@ -30,6 +30,7 @@ import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.aggregation.Aggregations;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple4;
+import org.apache.flink.api.java.utils.ParameterTool;
/**
* This program implements a modified version of the TPC-H query 3. The
@@ -71,7 +72,7 @@ import org.apache.flink.api.java.tuple.Tuple4;
* as generated by the TPC-H data generator which is available at <a href="http://www.tpc.org/tpch/">http://www.tpc.org/tpch/</a>.
*
* <p>
- * Usage: <code>TPCHQuery3 <lineitem-csv path> <customer-csv path> <orders-csv path> <result path></code><br>
+ * Usage: <code>TPCHQuery3 --lineitem<path> --customer <path> --orders<path> --output <path></code><br>
*
* <p>
* This example shows how to use:
@@ -90,16 +91,24 @@ public class TPCHQuery3 {
public static void main(String[] args) throws Exception {
- if(!parseParameters(args)) {
+ final ParameterTool params = ParameterTool.fromArgs(args);
+
+ if (!params.has("lineitem") && !params.has("customer") && !params.has("orders")) {
+ System.err.println(" This program expects data from the TPC-H benchmark as input data.");
+ System.err.println(" Due to legal restrictions, we can not ship generated data.");
+ System.out.println(" You can find the TPC-H data generator at http://www.tpc.org/tpch/.");
+ System.out.println(" Usage: TPCHQuery3 --lineitem <path> --customer <path> --orders <path> [--output <path>]");
return;
}
final ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();
+ env.getConfig().setGlobalJobParameters(params);
+
// get input data
- DataSet<Lineitem> lineitems = getLineitemDataSet(env);
- DataSet<Order> orders = getOrdersDataSet(env);
- DataSet<Customer> customers = getCustomerDataSet(env);
+ DataSet<Lineitem> lineitems = getLineitemDataSet(env, params.get("lineitem"));
+ DataSet<Order> orders = getOrdersDataSet(env, params.get("customer"));
+ DataSet<Customer> customers = getCustomerDataSet(env, params.get("orders"));
// Filter market segment "AUTOMOBILE"
customers = customers.filter(
@@ -162,11 +171,15 @@ public class TPCHQuery3 {
.aggregate(Aggregations.SUM, 1);
// emit result
- result.writeAsCsv(outputPath, "\n", "|");
-
- // execute program
- env.execute("TPCH Query 3 Example");
-
+ if (params.has("output")) {
+ result.writeAsCsv(params.get("output"), "\n", "|");
+ // execute program
+ env.execute("TPCH Query 3 Example");
+ } else {
+ System.out.println("Printing result to stdout. Use --output to specify output path.");
+ result.print();
+ }
+
}
// *************************************************************************
@@ -220,48 +233,21 @@ public class TPCHQuery3 {
// UTIL METHODS
// *************************************************************************
- private static String lineitemPath;
- private static String customerPath;
- private static String ordersPath;
- private static String outputPath;
-
- private static boolean parseParameters(String[] programArguments) {
-
- if(programArguments.length > 0) {
- if(programArguments.length == 4) {
- lineitemPath = programArguments[0];
- customerPath = programArguments[1];
- ordersPath = programArguments[2];
- outputPath = programArguments[3];
- } else {
- System.err.println("Usage: TPCHQuery3 <lineitem-csv path> <customer-csv path> <orders-csv path> <result path>");
- return false;
- }
- } else {
- System.err.println("This program expects data from the TPC-H benchmark as input data.\n" +
- " Due to legal restrictions, we can not ship generated data.\n" +
- " You can find the TPC-H data generator at http://www.tpc.org/tpch/.\n" +
- " Usage: TPCHQuery3 <lineitem-csv path> <customer-csv path> <orders-csv path> <result path>");
- return false;
- }
- return true;
- }
-
- private static DataSet<Lineitem> getLineitemDataSet(ExecutionEnvironment env) {
+ private static DataSet<Lineitem> getLineitemDataSet(ExecutionEnvironment env, String lineitemPath) {
return env.readCsvFile(lineitemPath)
.fieldDelimiter("|")
.includeFields("1000011000100000")
.tupleType(Lineitem.class);
}
- private static DataSet<Customer> getCustomerDataSet(ExecutionEnvironment env) {
+ private static DataSet<Customer> getCustomerDataSet(ExecutionEnvironment env, String customerPath) {
return env.readCsvFile(customerPath)
.fieldDelimiter("|")
.includeFields("10000010")
.tupleType(Customer.class);
}
- private static DataSet<Order> getOrdersDataSet(ExecutionEnvironment env) {
+ private static DataSet<Order> getOrdersDataSet(ExecutionEnvironment env, String ordersPath) {
return env.readCsvFile(ordersPath)
.fieldDelimiter("|")
.includeFields("110010010")