You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by fh...@apache.org on 2015/12/21 18:44:11 UTC
[8/8] flink git commit: [FLINK-2622] [streaming] Align Scala
streaming writeAsCsv API call with Java API
[FLINK-2622] [streaming] Align Scala streaming writeAsCsv API call with Java API
This closes #1473
This closes #1098
This closes #1459
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/c08bcf1e
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/c08bcf1e
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/c08bcf1e
Branch: refs/heads/master
Commit: c08bcf1e008397bfe22353e3ba3d9d9656e7d3c1
Parents: af99630
Author: Till Rohrmann <tr...@apache.org>
Authored: Fri Sep 18 15:31:29 2015 +0800
Committer: Fabian Hueske <fh...@apache.org>
Committed: Mon Dec 21 16:59:03 2015 +0100
----------------------------------------------------------------------
.../org/apache/flink/core/fs/FileSystem.java | 2 +-
.../streaming/api/datastream/DataStream.java | 198 ++++++++++---------
.../flink/streaming/api/scala/DataStream.scala | 182 ++++++++++++++---
.../scala/api/CsvOutputFormatITCase.java | 91 ++++++++-
.../scala/api/TextOutputFormatITCase.java | 74 ++++++-
.../api/scala/OutputFormatTestPrograms.scala | 127 ++++++++++--
6 files changed, 519 insertions(+), 155 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
----------------------------------------------------------------------
diff --git a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
index 03ca6a2..ae44213 100644
--- a/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
+++ b/flink-core/src/main/java/org/apache/flink/core/fs/FileSystem.java
@@ -64,7 +64,7 @@ public abstract class FileSystem {
/** Creates write path if it does not exist. Does not overwrite existing files and directories. */
NO_OVERWRITE,
- /** creates write path if it does not exist. Overwrites existing files and directories. */
+ /** Creates write path if it does not exist. Overwrites existing files and directories. */
OVERWRITE
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
----------------------------------------------------------------------
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
index b78689c..7ab3e2b 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/datastream/DataStream.java
@@ -92,7 +92,7 @@ import com.google.common.base.Preconditions;
* <li>{@link DataStream#map},
* <li>{@link DataStream#filter}, or
* </ul>
- *
+ *
* @param <T> The type of the elements in this Stream
*/
public class DataStream<T> {
@@ -114,7 +114,7 @@ public class DataStream<T> {
/**
* Returns the ID of the {@link DataStream} in the current {@link StreamExecutionEnvironment}.
- *
+ *
* @return ID of the DataStream
*/
public Integer getId() {
@@ -123,7 +123,7 @@ public class DataStream<T> {
/**
* Gets the parallelism for this operator.
- *
+ *
* @return The parallelism set for this operator.
*/
public int getParallelism() {
@@ -132,7 +132,7 @@ public class DataStream<T> {
/**
* Gets the type of the stream.
- *
+ *
* @return The type of the datastream.
*/
public TypeInformation<T> getType() {
@@ -167,7 +167,7 @@ public class DataStream<T> {
* Creates a new {@link DataStream} by merging {@link DataStream} outputs of
* the same type with each other. The DataStreams merged using this operator
* will be transformed simultaneously.
- *
+ *
* @param streams
* The DataStreams to union output with.
* @return The {@link DataStream}.
@@ -194,7 +194,7 @@ public class DataStream<T> {
* Operator used for directing tuples to specific named outputs using an
* {@link org.apache.flink.streaming.api.collector.selector.OutputSelector}.
* Calling this method on an operator creates a new {@link SplitStream}.
- *
+ *
* @param outputSelector
* The user defined
* {@link org.apache.flink.streaming.api.collector.selector.OutputSelector}
@@ -210,7 +210,7 @@ public class DataStream<T> {
* {@link DataStream} outputs of (possible) different types with each other.
* The DataStreams connected using this operator can be used with
* CoFunctions to apply joint transformations.
- *
+ *
* @param dataStream
* The DataStream with which this stream will be connected.
* @return The {@link ConnectedStreams}.
@@ -220,7 +220,7 @@ public class DataStream<T> {
}
/**
- *
+ *
* It creates a new {@link KeyedStream} that uses the provided key for partitioning
* its operator states.
*
@@ -391,7 +391,7 @@ public class DataStream<T> {
* <p>
* This setting only effects the how the outputs will be distributed between
* the parallel instances of the next processing operator.
- *
+ *
* @return The DataStream with broadcast partitioning set.
*/
public DataStream<T> broadcast() {
@@ -405,7 +405,7 @@ public class DataStream<T> {
* <p>
* This setting only effects the how the outputs will be distributed between
* the parallel instances of the next processing operator.
- *
+ *
* @return The DataStream with shuffle partitioning set.
*/
public DataStream<T> shuffle() {
@@ -420,7 +420,7 @@ public class DataStream<T> {
* <p>
* This setting only effects the how the outputs will be distributed between
* the parallel instances of the next processing operator.
- *
+ *
* @return The DataStream with forward partitioning set.
*/
public DataStream<T> forward() {
@@ -435,7 +435,7 @@ public class DataStream<T> {
* <p>
* This setting only effects the how the outputs will be distributed between
* the parallel instances of the next processing operator.
- *
+ *
* @return The DataStream with rebalance partitioning set.
*/
public DataStream<T> rebalance() {
@@ -447,7 +447,7 @@ public class DataStream<T> {
* all go to the first instance of the next processing operator. Use this
* setting with care since it might cause a serious performance bottleneck
* in the application.
- *
+ *
* @return The DataStream with shuffle partitioning set.
*/
public DataStream<T> global() {
@@ -478,7 +478,7 @@ public class DataStream<T> {
* can use the maxWaitTime parameter to set a max waiting time for the
* iteration head. If no data received in the set time, the stream
* terminates.
- *
+ *
* @return The iterative data stream created.
*/
public IterativeStream<T> iterate() {
@@ -509,11 +509,11 @@ public class DataStream<T> {
* can use the maxWaitTime parameter to set a max waiting time for the
* iteration head. If no data received in the set time, the stream
* terminates.
- *
+ *
* @param maxWaitTimeMillis
* Number of milliseconds to wait between inputs before shutting
* down
- *
+ *
* @return The iterative data stream created.
*/
public IterativeStream<T> iterate(long maxWaitTimeMillis) {
@@ -526,7 +526,7 @@ public class DataStream<T> {
* MapFunction call returns exactly one element. The user can also extend
* {@link RichMapFunction} to gain access to other features provided by the
* {@link org.apache.flink.api.common.functions.RichFunction} interface.
- *
+ *
* @param mapper
* The MapFunction that is called for each element of the
* DataStream.
@@ -549,11 +549,11 @@ public class DataStream<T> {
* including none. The user can also extend {@link RichFlatMapFunction} to
* gain access to other features provided by the
* {@link org.apache.flink.api.common.functions.RichFunction} interface.
- *
+ *
* @param flatMapper
* The FlatMapFunction that is called for each element of the
* DataStream
- *
+ *
* @param <R>
* output type
* @return The transformed {@link DataStream}.
@@ -575,7 +575,7 @@ public class DataStream<T> {
* user can also extend {@link RichFilterFunction} to gain access to other
* features provided by the
* {@link org.apache.flink.api.common.functions.RichFunction} interface.
- *
+ *
* @param filter
* The FilterFunction that is called for each element of the
* DataStream.
@@ -593,13 +593,13 @@ public class DataStream<T> {
* <p>
* The transformation projects each Tuple of the DataSet onto a (sub)set of
* fields.
- *
+ *
* @param fieldIndexes
* The field indexes of the input tuples that are retained. The
* order of fields in the output tuple corresponds to the order
* of field indexes.
* @return The projected DataStream
- *
+ *
* @see Tuple
* @see DataStream
*/
@@ -748,7 +748,7 @@ public class DataStream<T> {
* <p>
* For each element of the DataStream the result of
* {@link Object#toString()} is written.
- *
+ *
* @return The closed DataStream.
*/
public DataStreamSink<T> print() {
@@ -762,7 +762,7 @@ public class DataStream<T> {
* <p>
* For each element of the DataStream the result of
* {@link Object#toString()} is written.
- *
+ *
* @return The closed DataStream.
*/
public DataStreamSink<T> printToErr() {
@@ -776,11 +776,11 @@ public class DataStream<T> {
* <p>
* For every element of the DataStream the result of {@link Object#toString()}
* is written.
- *
+ *
* @param path
- * the path pointing to the location the text file is written to
- *
- * @return the closed DataStream.
+ * The path pointing to the location the text file is written to.
+ *
+ * @return The closed DataStream.
*/
public DataStreamSink<T> writeAsText(String path) {
return write(new TextOutputFormat<T>(new Path(path)), 0L);
@@ -793,13 +793,13 @@ public class DataStream<T> {
* <p>
* For every element of the DataStream the result of {@link Object#toString()}
* is written.
- *
+ *
* @param path
- * the path pointing to the location the text file is written to
+ * The path pointing to the location the text file is written to.
* @param millis
- * the file update frequency
- *
- * @return the closed DataStream
+ * The file update frequency.
+ *
+ * @return The closed DataStream.
*/
public DataStreamSink<T> writeAsText(String path, long millis) {
TextOutputFormat<T> tof = new TextOutputFormat<T>(new Path(path));
@@ -812,14 +812,14 @@ public class DataStream<T> {
* <p>
* For every element of the DataStream the result of {@link Object#toString()}
* is written.
- *
+ *
* @param path
- * the path pointing to the location the text file is written to
+ * The path pointing to the location the text file is written to
* @param writeMode
- * Control the behavior for existing files. Options are
+ * Controls the behavior for existing files. Options are
* NO_OVERWRITE and OVERWRITE.
- *
- * @return the closed DataStream.
+ *
+ * @return The closed DataStream.
*/
public DataStreamSink<T> writeAsText(String path, WriteMode writeMode) {
TextOutputFormat<T> tof = new TextOutputFormat<T>(new Path(path));
@@ -833,16 +833,16 @@ public class DataStream<T> {
* <p>
* For every element of the DataStream the result of {@link Object#toString()}
* is written.
- *
+ *
* @param path
- * the path pointing to the location the text file is written to
+ * The path pointing to the location the text file is written to
* @param writeMode
* Controls the behavior for existing files. Options are
* NO_OVERWRITE and OVERWRITE.
* @param millis
- * the file update frequency
- *
- * @return the closed DataStream.
+ T the file update frequency
+ *
+ * @return The closed DataStream.
*/
public DataStreamSink<T> writeAsText(String path, WriteMode writeMode, long millis) {
TextOutputFormat<T> tof = new TextOutputFormat<T>(new Path(path));
@@ -851,28 +851,23 @@ public class DataStream<T> {
}
/**
- * Writes a DataStream to the file specified by path in csv format.
+ * Writes a DataStream to the file specified by the path parameter.
*
* <p>
* For every field of an element of the DataStream the result of {@link Object#toString()}
* is written. This method can only be used on data streams of tuples.
- *
+ *
* @param path
* the path pointing to the location the text file is written to
- *
+ *
* @return the closed DataStream
*/
- @SuppressWarnings("unchecked")
- public <X extends Tuple> DataStreamSink<T> writeAsCsv(String path) {
- Preconditions.checkArgument(getType().isTupleType(),
- "The writeAsCsv() method can only be used on data sets of tuples.");
- CsvOutputFormat<X> of = new CsvOutputFormat<X>(new Path(path),
- CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
- return write((OutputFormat<T>) of, 0L);
+ public DataStreamSink<T> writeAsCsv(String path) {
+ return writeAsCsv(path, null, 0L, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
}
/**
- * Writes a DataStream to the file specified by path in csv format. The
+ * Writes a DataStream to the file specified by the path parameter. The
* writing is performed periodically, in every millis milliseconds.
*
* <p>
@@ -883,53 +878,62 @@ public class DataStream<T> {
* the path pointing to the location the text file is written to
* @param millis
* the file update frequency
- *
+ *
* @return the closed DataStream
*/
- @SuppressWarnings("unchecked")
- public <X extends Tuple> DataStreamSink<T> writeAsCsv(String path, long millis) {
- Preconditions.checkArgument(getType().isTupleType(),
- "The writeAsCsv() method can only be used on data sets of tuples.");
- CsvOutputFormat<X> of = new CsvOutputFormat<X>(new Path(path),
- CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
- return write((OutputFormat<T>) of, millis);
+ public DataStreamSink<T> writeAsCsv(String path, long millis) {
+ return writeAsCsv(path, null, millis, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
}
/**
- * Writes a DataStream to the file specified by path in csv format.
+ * Writes a DataStream to the file specified by the path parameter.
*
* <p>
* For every field of an element of the DataStream the result of {@link Object#toString()}
* is written. This method can only be used on data streams of tuples.
- *
+ *
* @param path
* the path pointing to the location the text file is written to
* @param writeMode
* Controls the behavior for existing files. Options are
* NO_OVERWRITE and OVERWRITE.
- *
+ *
* @return the closed DataStream
*/
- @SuppressWarnings("unchecked")
- public <X extends Tuple> DataStreamSink<T> writeAsCsv(String path, WriteMode writeMode) {
- Preconditions.checkArgument(getType().isTupleType(),
- "The writeAsCsv() method can only be used on data sets of tuples.");
- CsvOutputFormat<X> of = new CsvOutputFormat<X>(new Path(path),
- CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
- if (writeMode != null) {
- of.setWriteMode(writeMode);
- }
- return write((OutputFormat<T>) of, 0L);
+ public DataStreamSink<T> writeAsCsv(String path, WriteMode writeMode) {
+ return writeAsCsv(path, writeMode, 0L, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
}
/**
- * Writes a DataStream to the file specified by path in csv format. The
- * writing is performed periodically, in every millis milliseconds.
+ * Writes a DataStream to the file specified by the path parameter. The
+ * writing is performed periodically every millis milliseconds.
+ *
+ * <p>
+ * For every field of an element of the DataStream the result of {@link Object#toString()}
+ * is written. This method can only be used on data streams of tuples.
+ *
+ * @param path
+ * the path pointing to the location the text file is written to
+ * @param writeMode
+ * Controls the behavior for existing files. Options are
+ * NO_OVERWRITE and OVERWRITE.
+ * @param millis
+ * the file update frequency
+ *
+ * @return the closed DataStream
+ */
+ public DataStreamSink<T> writeAsCsv(String path, WriteMode writeMode, long millis) {
+ return writeAsCsv(path, writeMode, millis, CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
+ }
+
+ /**
+ * Writes a DataStream to the file specified by the path parameter. The
+ * writing is performed periodically every millis milliseconds.
*
* <p>
* For every field of an element of the DataStream the result of {@link Object#toString()}
* is written. This method can only be used on data streams of tuples.
- *
+ *
* @param path
* the path pointing to the location the text file is written to
* @param writeMode
@@ -937,26 +941,40 @@ public class DataStream<T> {
* NO_OVERWRITE and OVERWRITE.
* @param millis
* the file update frequency
- *
+ * @param rowDelimiter
+ * the delimiter for two rows
+ * @param fieldDelimiter
+ * the delimiter for two fields
+ *
* @return the closed DataStream
*/
@SuppressWarnings("unchecked")
- public <X extends Tuple> DataStreamSink<T> writeAsCsv(String path, WriteMode writeMode,
- long millis) {
- Preconditions.checkArgument(getType().isTupleType(),
- "The writeAsCsv() method can only be used on data sets of tuples.");
- CsvOutputFormat<X> of = new CsvOutputFormat<X>(new Path(path),
- CsvOutputFormat.DEFAULT_LINE_DELIMITER, CsvOutputFormat.DEFAULT_FIELD_DELIMITER);
+ public <X extends Tuple> DataStreamSink<T> writeAsCsv(
+ String path,
+ WriteMode writeMode,
+ long millis,
+ String rowDelimiter,
+ String fieldDelimiter) {
+ Preconditions.checkArgument(
+ getType().isTupleType(),
+ "The writeAsCsv() method can only be used on data streams of tuples.");
+
+ CsvOutputFormat<X> of = new CsvOutputFormat<X>(
+ new Path(path),
+ rowDelimiter,
+ fieldDelimiter);
+
if (writeMode != null) {
of.setWriteMode(writeMode);
}
+
return write((OutputFormat<T>) of, millis);
}
/**
* Writes the DataStream to a socket as a byte array. The format of the
* output is specified by a {@link SerializationSchema}.
- *
+ *
* @param hostName
* host of the socket
* @param port
@@ -970,10 +988,10 @@ public class DataStream<T> {
returnStream.setParallelism(1); // It would not work if multiple instances would connect to the same port
return returnStream;
}
-
+
/**
* Writes the dataStream into an output, described by an OutputFormat.
- *
+ *
* @param format The output format
* @param millis the write frequency
* @return The closed DataStream
@@ -985,7 +1003,7 @@ public class DataStream<T> {
/**
* Method for passing user defined operators along with the type
* information that will transform the DataStream.
- *
+ *
* @param operatorName
* name of the operator, for logging purposes
* @param outTypeInfo
@@ -1031,7 +1049,7 @@ public class DataStream<T> {
* Adds the given sink to this DataStream. Only streams with sinks added
* will be executed once the {@link StreamExecutionEnvironment#execute()}
* method is called.
- *
+ *
* @param sinkFunction
* The object containing the sink's invoke function.
* @return The closed DataStream.
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
index e9cc041..10c2daf 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/DataStream.scala
@@ -76,7 +76,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
* Returns the parallelism of this operation.
*/
def getParallelism = javaStream.getParallelism
-
+
/**
* Returns the execution config.
*/
@@ -105,13 +105,13 @@ class DataStream[T](javaStream: JavaStream[T]) {
case _ => throw new UnsupportedOperationException("Only supported for operators.")
this
}
-
+
/**
* Turns off chaining for this operator so thread co-location will not be
* used as an optimization. </p> Chaining can be turned off for the whole
* job by [[StreamExecutionEnvironment.disableOperatorChaining()]]
* however it is not advised for performance considerations.
- *
+ *
*/
def disableChaining(): DataStream[T] = {
javaStream match {
@@ -121,12 +121,12 @@ class DataStream[T](javaStream: JavaStream[T]) {
}
this
}
-
+
/**
* Starts a new task chain beginning at this operator. This operator will
* not be chained (thread co-located for increased performance) to any
* previous tasks even if possible.
- *
+ *
*/
def startNewChain(): DataStream[T] = {
javaStream match {
@@ -136,13 +136,13 @@ class DataStream[T](javaStream: JavaStream[T]) {
}
this
}
-
+
/**
* Isolates the operator in its own resource group. This will cause the
* operator to grab as many task slots as its degree of parallelism. If
* there are no free resources available, the job will fail to start.
* All subsequent operators are assigned to the default resource group.
- *
+ *
*/
def isolateResources(): DataStream[T] = {
javaStream match {
@@ -152,7 +152,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
}
this
}
-
+
/**
* By default all operators in a streaming job share the same resource
* group. Each resource group takes as many task manager slots as the
@@ -374,19 +374,19 @@ class DataStream[T](javaStream: JavaStream[T]) {
iterativeStream.closeWith(feedback.getJavaStream)
output
}
-
+
/**
* Initiates an iterative part of the program that creates a loop by feeding
* back data streams. To create a streaming iteration the user needs to define
* a transformation that creates two DataStreams. The first one is the output
* that will be fed back to the start of the iteration and the second is the output
* stream of the iterative part.
- *
+ *
* The input stream of the iterate operator and the feedback stream will be treated
* as a ConnectedStreams where the the input is connected with the feedback stream.
- *
+ *
* This allows the user to distinguish standard input from feedback inputs.
- *
+ *
* <p>
* stepfunction: initialStream => (feedback, output)
* <p>
@@ -404,7 +404,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
val (feedback, output) = stepFunction(connectedIterativeStream)
connectedIterativeStream.closeWith(feedback.getJavaStream)
output
- }
+ }
/**
* Creates a new DataStream by applying the given function to every element of this DataStream.
@@ -417,7 +417,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
val mapper = new MapFunction[T, R] {
def map(in: T): R = cleanFun(in)
}
-
+
map(mapper)
}
@@ -441,7 +441,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
if (flatMapper == null) {
throw new NullPointerException("FlatMap function must not be null.")
}
-
+
val outType : TypeInformation[R] = implicitly[TypeInformation[R]]
javaStream.flatMap(flatMapper).returns(outType).asInstanceOf[JavaStream[R]]
}
@@ -671,7 +671,7 @@ class DataStream[T](javaStream: JavaStream[T]) {
/**
* Writes a DataStream to the standard output stream (stderr).
- *
+ *
* For each element of the DataStream the result of
* [[AnyRef.toString()]] is written.
*
@@ -680,28 +680,154 @@ class DataStream[T](javaStream: JavaStream[T]) {
def printToErr() = javaStream.printToErr()
/**
+ * Writes a DataStream to the file specified by path in text format. For
+ * every element of the DataStream the result of .toString is written.
+ *
+ * @param path The path pointing to the location the text file is written to
+ * @return The closed DataStream
+ */
+ def writeAsText(path: String): DataStreamSink[T] =
+ javaStream.writeAsText(path, 0L)
+
+ /**
* Writes a DataStream to the file specified by path in text format. The
- * writing is performed periodically, in every millis milliseconds. For
+ * writing is performed periodically, every millis milliseconds. For
* every element of the DataStream the result of .toString
* is written.
*
+ * @param path The path pointing to the location the text file is written to
+ * @param millis The file update frequency
+ * @return The closed DataStream
*/
- def writeAsText(path: String, millis: Long = 0): DataStreamSink[T] =
+ def writeAsText(path: String, millis: Long): DataStreamSink[T] =
javaStream.writeAsText(path, millis)
/**
- * Writes a DataStream to the file specified by path in text format. The
- * writing is performed periodically, in every millis milliseconds. For
- * every element of the DataStream the result of .toString
- * is written.
- *
- */
+ * Writes a DataStream to the file specified by path in text format. For
+ * every element of the DataStream the result of .toString is written.
+ *
+ * @param path The path pointing to the location the text file is written to
+ * @param writeMode Controls the behavior for existing files. Options are NO_OVERWRITE and
+ * OVERWRITE.
+ * @return The closed DataStream
+ */
+ def writeAsText(path: String, writeMode: FileSystem.WriteMode): DataStreamSink[T] = {
+ if (writeMode != null) {
+ javaStream.writeAsText(path, writeMode)
+ } else {
+ javaStream.writeAsText(path)
+ }
+ }
+
+ /**
+ * Writes a DataStream to the file specified by path in text format. The writing is performed
+ * periodically every millis milliseconds. For every element of the DataStream the result of
+ * .toString is written.
+ *
+ * @param path The path pointing to the location the text file is written to
+ * @param writeMode Controls the behavior for existing files. Options are NO_OVERWRITE and
+ * OVERWRITE.
+ * @param millis The file update frequency
+ * @return The closed DataStream
+ */
+ def writeAsText(
+ path: String,
+ writeMode: FileSystem.WriteMode,
+ millis: Long)
+ : DataStreamSink[T] = {
+ if (writeMode != null) {
+ javaStream.writeAsText(path, writeMode, millis)
+ } else {
+ javaStream.writeAsText(path, millis)
+ }
+ }
+
+ /**
+ * Writes the DataStream in CSV format to the file specified by the path parameter. The writing
+ * is performed periodically every millis milliseconds.
+ *
+ * @param path Path to the location of the CSV file
+ * @return The closed DataStream
+ */
+ def writeAsCsv(path: String): DataStreamSink[T] = {
+ writeAsCsv(
+ path,
+ null,
+ 0L,
+ ScalaCsvOutputFormat.DEFAULT_LINE_DELIMITER,
+ ScalaCsvOutputFormat.DEFAULT_FIELD_DELIMITER)
+ }
+
+ /**
+ * Writes the DataStream in CSV format to the file specified by the path parameter. The writing
+ * is performed periodically every millis milliseconds.
+ *
+ * @param path Path to the location of the CSV file
+ * @param millis File update frequency
+ * @return The closed DataStream
+ */
+ def writeAsCsv(path: String, millis: Long): DataStreamSink[T] = {
+ writeAsCsv(
+ path,
+ null,
+ millis,
+ ScalaCsvOutputFormat.DEFAULT_LINE_DELIMITER,
+ ScalaCsvOutputFormat.DEFAULT_FIELD_DELIMITER)
+ }
+
+ /**
+ * Writes the DataStream in CSV format to the file specified by the path parameter. The writing
+ * is performed periodically every millis milliseconds.
+ *
+ * @param path Path to the location of the CSV file
+ * @param writeMode Controls whether an existing file is overwritten or not
+ * @return The closed DataStream
+ */
+ def writeAsCsv(path: String, writeMode: FileSystem.WriteMode): DataStreamSink[T] = {
+ writeAsCsv(
+ path,
+ writeMode,
+ 0L,
+ ScalaCsvOutputFormat.DEFAULT_LINE_DELIMITER,
+ ScalaCsvOutputFormat.DEFAULT_FIELD_DELIMITER)
+ }
+
+ /**
+ * Writes the DataStream in CSV format to the file specified by the path parameter. The writing
+ * is performed periodically every millis milliseconds.
+ *
+ * @param path Path to the location of the CSV file
+ * @param writeMode Controls whether an existing file is overwritten or not
+ * @param millis File update frequency
+ * @return The closed DataStream
+ */
+ def writeAsCsv(path: String, writeMode: FileSystem.WriteMode, millis: Long): DataStreamSink[T] = {
+ writeAsCsv(
+ path,
+ writeMode,
+ millis,
+ ScalaCsvOutputFormat.DEFAULT_LINE_DELIMITER,
+ ScalaCsvOutputFormat.DEFAULT_FIELD_DELIMITER)
+ }
+
+ /**
+ * Writes the DataStream in CSV format to the file specified by the path parameter. The writing
+ * is performed periodically every millis milliseconds.
+ *
+ * @param path Path to the location of the CSV file
+ * @param writeMode Controls whether an existing file is overwritten or not
+ * @param millis File update frequency
+ * @param rowDelimiter Delimiter for consecutive rows
+ * @param fieldDelimiter Delimiter for consecutive fields
+ * @return The closed DataStream
+ */
def writeAsCsv(
path: String,
- millis: Long = 0,
- rowDelimiter: String = ScalaCsvOutputFormat.DEFAULT_LINE_DELIMITER,
- fieldDelimiter: String = ScalaCsvOutputFormat.DEFAULT_FIELD_DELIMITER,
- writeMode: FileSystem.WriteMode = null): DataStreamSink[T] = {
+ writeMode: FileSystem.WriteMode,
+ millis: Long,
+ rowDelimiter: String,
+ fieldDelimiter: String)
+ : DataStreamSink[T] = {
require(javaStream.getType.isTupleType, "CSV output can only be used with Tuple DataSets.")
val of = new ScalaCsvOutputFormat[Product](new Path(path), rowDelimiter, fieldDelimiter)
if (writeMode != null) {
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/CsvOutputFormatITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/CsvOutputFormatITCase.java b/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/CsvOutputFormatITCase.java
index a389e51..10dd5d3 100644
--- a/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/CsvOutputFormatITCase.java
+++ b/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/CsvOutputFormatITCase.java
@@ -19,30 +19,101 @@ package org.apache.flink.streaming.scala.api;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.scala.OutputFormatTestPrograms;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.AbstractTestBase;
import org.apache.flink.util.Collector;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
-public class CsvOutputFormatITCase extends StreamingProgramTestBase {
+import java.io.File;
+import static org.junit.Assert.*;
+
+
+public class CsvOutputFormatITCase extends StreamingMultipleProgramsTestBase {
protected String resultPath;
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public AbstractTestBase fileInfo = new AbstractTestBase(new Configuration()) {
+ @Override
+ public void startCluster() throws Exception {
+ super.startCluster();
+ }
+ };
+
+ @Before
+ public void createFile() throws Exception {
+ File f = fileInfo.createAndRegisterTempFile("result");
+ resultPath = f.toURI().toString();
}
- @Override
- protected void testProgram() throws Exception {
+ @Test
+ public void testPath() throws Exception {
OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath);
}
- @Override
- protected void postSubmit() throws Exception {
- //Strip the parentheses from the expected text like output
+ @Test
+ public void testPathMillis() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, 1);
+ }
+
+ @Test
+ public void testPathWriteMode() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE);
+ }
+
+ @Test
+ public void testPathWriteModeMillis() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1);
+ }
+
+ @Test
+ public void testPathWriteModeMillisDelimiter() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1, "\n", ",");
+ }
+
+ @Test
+ public void failPathWriteMode() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath);
+ try {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE);
+ fail("File should exist.");
+ } catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("File already exists"));
+ }
+ }
+
+ @Test
+ public void failPathWriteModeMillis() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath);
+ try {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1);
+ fail("File should exist");
+ } catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("File already exists"));
+ }
+ }
+
+ @Test
+ public void failPathWriteModeMillisDelimiter() throws Exception {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath);
+ try {
+ OutputFormatTestPrograms.wordCountToCsv(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1, "\n", ",");
+ fail("File should exist.");
+ } catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("File already exists"));
+ }
+ }
+
+ @After
+ public void closeFile() throws Exception {
compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES
.replaceAll("[\\\\(\\\\)]", ""), resultPath);
+ fileInfo.stopCluster();
}
public static final class Tokenizer implements FlatMapFunction<String, Tuple2<String, Integer>> {
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/TextOutputFormatITCase.java
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/TextOutputFormatITCase.java b/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/TextOutputFormatITCase.java
index 530ba67..d2cc6db 100644
--- a/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/TextOutputFormatITCase.java
+++ b/flink-streaming-scala/src/test/java/org/apache/flink/streaming/scala/api/TextOutputFormatITCase.java
@@ -17,27 +17,83 @@
package org.apache.flink.streaming.scala.api;
+import org.apache.flink.configuration.Configuration;
+import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.streaming.api.scala.OutputFormatTestPrograms;
-import org.apache.flink.streaming.util.StreamingProgramTestBase;
+import org.apache.flink.streaming.util.StreamingMultipleProgramsTestBase;
import org.apache.flink.test.testdata.WordCountData;
+import org.apache.flink.test.util.AbstractTestBase;
+import org.junit.After;
+import org.junit.Before;
+import org.junit.Test;
-public class TextOutputFormatITCase extends StreamingProgramTestBase {
+import java.io.File;
+
+import static org.junit.Assert.*;
+
+public class TextOutputFormatITCase extends StreamingMultipleProgramsTestBase {
protected String resultPath;
- @Override
- protected void preSubmit() throws Exception {
- resultPath = getTempDirPath("result");
+ public AbstractTestBase fileInfo = new AbstractTestBase(new Configuration()) {
+ @Override
+ public void startCluster() throws Exception {
+ super.startCluster();
+ }
+ };
+
+ @Before
+ public void createFile() throws Exception {
+ File f = fileInfo.createAndRegisterTempFile("result");
+ resultPath = f.toURI().toString();
+ }
+
+ @Test
+ public void testPath() throws Exception {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath);
+ }
+
+ @Test
+ public void testPathMillis() throws Exception {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath, 1);
+ }
+
+ @Test
+ public void testPathWriteMode() throws Exception {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE);
+ }
+
+ @Test
+ public void testPathWriteModeMillis() throws Exception {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1);
+ }
+
+ @Test
+ public void failPathWriteMode() throws Exception {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath);
+ try {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE);
+ fail("File should exist.");
+ } catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("File already exists"));
+ }
}
- @Override
- protected void testProgram() throws Exception {
+ @Test
+ public void failPathWriteModeMillis() throws Exception {
OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath);
+ try {
+ OutputFormatTestPrograms.wordCountToText(WordCountData.TEXT, resultPath, FileSystem.WriteMode.NO_OVERWRITE, 1);
+ fail("File should exist.");
+ } catch (Exception e) {
+ assertTrue(e.getCause().getMessage().contains("File already exists"));
+ }
}
- @Override
- protected void postSubmit() throws Exception {
+ @After
+ public void closeFile() throws Exception {
compareResultsByLinesInMemory(WordCountData.STREAMING_COUNTS_AS_TUPLES, resultPath);
+ fileInfo.stopCluster();
}
}
http://git-wip-us.apache.org/repos/asf/flink/blob/c08bcf1e/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
index 36b1068..e0a9a9c 100644
--- a/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
+++ b/flink-streaming-scala/src/test/scala/org/apache/flink/streaming/api/scala/OutputFormatTestPrograms.scala
@@ -18,54 +18,147 @@
package org.apache.flink.streaming.api.scala
+import org.apache.flink.core.fs.FileSystem
import org.apache.flink.streaming.util.serialization.SimpleStringSchema
import scala.language.existentials
/**
- * Test programs for built in output formats. Invoked from {@link OutputFormatTest}.
+ * Test programs for built in output formats. Invoked from OutputFormatTest.
*/
object OutputFormatTestPrograms {
- def wordCountToText(input : String, outputPath : String) : Unit = {
- val env = StreamExecutionEnvironment.getExecutionEnvironment
-
- //Create streams for names and ages by mapping the inputs to the corresponding objects
- val text = env.fromElements(input)
- val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
+ def wordCountProgram(input: DataStream[String]): DataStream[(String, Int)] = {
+ input.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
.map { (_, 1) }
.keyBy(0)
.sum(1)
+ }
+
+ def wordCountToText(input: String, outputPath : String) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
counts.writeAsText(outputPath)
env.execute("Scala WordCountToText")
}
- def wordCountToCsv(input : String, outputPath : String) : Unit = {
+ def wordCountToText(input : String, outputPath : String, millis : Long) : Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
- //Create streams for names and ages by mapping the inputs to the corresponding objects
+ val counts = wordCountProgram(text)
+ counts.writeAsText(outputPath, millis)
+
+ env.execute("Scala WordCountToText")
+ }
+
+ def wordCountToText(
+ input : String,
+ outputPath : String,
+ writeMode : FileSystem.WriteMode) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsText(outputPath, writeMode)
+
+ env.execute("Scala WordCountToText")
+ }
+
+ def wordCountToText(
+ input : String,
+ outputPath : String,
+ writeMode : FileSystem.WriteMode,
+ millis : Long) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsText(outputPath, writeMode, millis)
+
+ env.execute("Scala WordCountToText")
+ }
+
+ def wordCountToCsv(input : String, outputPath : String) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
val text = env.fromElements(input)
- val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
- .map { (_, 1) }
- .keyBy(0)
- .sum(1)
+
+ val counts = wordCountProgram(text)
counts.writeAsCsv(outputPath)
env.execute("Scala WordCountToCsv")
}
+ def wordCountToCsv(input : String, outputPath : String, millis : Long) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsCsv(outputPath, millis)
+
+ env.execute("Scala WordCountToCsv")
+ }
+
+ def wordCountToCsv(
+ input : String,
+ outputPath : String,
+ writeMode : FileSystem.WriteMode) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsCsv(outputPath, writeMode)
+
+ env.execute("Scala WordCountToCsv")
+ }
+
+ def wordCountToCsv(
+ input : String,
+ outputPath : String,
+ writeMode : FileSystem.WriteMode,
+ millis : Long) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsCsv(outputPath, writeMode, millis)
+
+ env.execute("Scala WordCountToCsv")
+ }
+
+ def wordCountToCsv(
+ input : String,
+ outputPath : String,
+ writeMode : FileSystem.WriteMode,
+ millis : Long,
+ rowDelimiter: String,
+ fieldDelimiter: String) : Unit = {
+ val env = StreamExecutionEnvironment.getExecutionEnvironment
+ val text = env.fromElements(input)
+
+ val counts = wordCountProgram(text)
+
+ counts.writeAsCsv(outputPath, writeMode, millis, rowDelimiter, fieldDelimiter)
+
+ env.execute("Scala WordCountToCsv")
+ }
+
def wordCountToSocket(input : String, outputHost : String, outputPort : Int) : Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
//Create streams for names and ages by mapping the inputs to the corresponding objects
val text = env.fromElements(input)
- val counts = text.flatMap { _.toLowerCase.split("\\W+") filter { _.nonEmpty } }
- .map { (_, 1) }
- .keyBy(0)
- .sum(1)
+ val counts = wordCountProgram(text)
.map(tuple => tuple.toString() + "\n")
counts.writeToSocket(outputHost, outputPort, new SimpleStringSchema())