flink-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From fhue...@apache.org
Subject [12/50] [abbrv] flink git commit: [FLINK-3366] Rename @Experimental annotation to @PublicEvolving
Date Fri, 12 Feb 2016 11:29:37 GMT
http://git-wip-us.apache.org/repos/asf/flink/blob/572855da/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 736c41b..04c1980 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
@@ -18,7 +18,7 @@
 
 package org.apache.flink.streaming.api.scala
 
-import org.apache.flink.annotation.{Experimental, Public}
+import org.apache.flink.annotation.{PublicEvolving, Public}
 import org.apache.flink.api.common.functions.{FilterFunction, FlatMapFunction, MapFunction,
Partitioner}
 import org.apache.flink.api.common.io.OutputFormat
 import org.apache.flink.api.common.typeinfo.TypeInformation
@@ -50,6 +50,7 @@ class DataStream[T](stream: JavaStream[T]) {
 
   /**
     * Returns the [[StreamExecutionEnvironment]] associated with the current [[DataStream]].
+ *
     * @return associated execution environment
     */
   def getExecutionEnvironment: StreamExecutionEnvironment =
@@ -60,7 +61,7 @@ class DataStream[T](stream: JavaStream[T]) {
    *
    * @return ID of the DataStream
    */
-  @Experimental
+  @PublicEvolving
   def getId = stream.getId
 
   /**
@@ -128,7 +129,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param uid The unique user-specified ID of this transformation.
     * @return The operator with the specified ID.
     */
-  @Experimental
+  @PublicEvolving
   def uid(uid: String) : DataStream[T] = javaStream match {
     case stream : SingleOutputStreamOperator[T,_] => stream.uid(uid)
     case _ => throw new UnsupportedOperationException("Only supported for operators.")
@@ -142,7 +143,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * however it is not advised for performance considerations.
    *
    */
-  @Experimental
+  @PublicEvolving
   def disableChaining(): DataStream[T] = {
     stream match {
       case ds: SingleOutputStreamOperator[_, _] => ds.disableChaining();
@@ -158,7 +159,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * previous tasks even if possible.
    *
    */
-  @Experimental
+  @PublicEvolving
   def startNewChain(): DataStream[T] = {
     stream match {
       case ds: SingleOutputStreamOperator[_, _] => ds.startNewChain();
@@ -175,7 +176,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * All subsequent operators are assigned to the default resource group.
    *
    */
-  @Experimental
+  @PublicEvolving
   def isolateResources(): DataStream[T] = {
     stream match {
       case ds: SingleOutputStreamOperator[_, _] => ds.isolateResources();
@@ -196,7 +197,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * degree of parallelism for the operators must be decreased from the
    * default.
    */
-  @Experimental
+  @PublicEvolving
   def startNewResourceGroup(): DataStream[T] = {
     stream match {
       case ds: SingleOutputStreamOperator[_, _] => ds.startNewResourceGroup();
@@ -345,14 +346,14 @@ class DataStream[T](stream: JavaStream[T]) {
    * the first instance of the next processing operator. Use this setting with care
    * since it might cause a serious performance bottleneck in the application.
    */
-  @Experimental
+  @PublicEvolving
   def global: DataStream[T] = stream.global()
 
   /**
    * Sets the partitioning of the DataStream so that the output tuples
    * are shuffled to the next component.
    */
-  @Experimental
+  @PublicEvolving
   def shuffle: DataStream[T] = stream.shuffle()
 
   /**
@@ -385,7 +386,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * In cases where the different parallelisms are not multiples of each other one or several
    * downstream operations will have a differing number of inputs from upstream operations.
    */
-  @Experimental
+  @PublicEvolving
   def rescale: DataStream[T] = stream.rescale()
 
   /**
@@ -408,7 +409,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * the keepPartitioning flag to true
    *
    */
-  @Experimental
+  @PublicEvolving
   def iterate[R](stepFunction: DataStream[T] => (DataStream[T], DataStream[R]),
                     maxWaitTimeMillis:Long = 0,
                     keepPartitioning: Boolean = false) : DataStream[R] = {
@@ -438,7 +439,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * to 0 then the iteration sources will indefinitely, so the job must be killed to stop.
    *
    */
-  @Experimental
+  @PublicEvolving
   def iterate[R, F: TypeInformation: ClassTag](stepFunction: ConnectedStreams[T, F] =>
     (DataStream[F], DataStream[R]), maxWaitTimeMillis:Long): DataStream[R] = {
     val feedbackType: TypeInformation[F] = implicitly[TypeInformation[F]]
@@ -625,7 +626,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * @param assigner The `WindowAssigner` that assigns elements to windows.
    * @return The trigger windows data stream.
    */
-  @Experimental
+  @PublicEvolving
   def windowAll[W <: Window](assigner: WindowAssigner[_ >: T, W]): AllWindowedStream[T,
W] = {
     new AllWindowedStream[T, W](new JavaAllWindowedStream[T, W](stream, assigner))
   }
@@ -640,7 +641,7 @@ class DataStream[T](stream: JavaStream[T]) {
    *
    * @see org.apache.flink.streaming.api.watermark.Watermark
    */
-  @Experimental
+  @PublicEvolving
   def assignTimestamps(extractor: TimestampExtractor[T]): DataStream[T] = {
     stream.assignTimestamps(clean(extractor))
   }
@@ -656,7 +657,7 @@ class DataStream[T](stream: JavaStream[T]) {
    *
    * @see org.apache.flink.streaming.api.watermark.Watermark
    */
-  @Experimental
+  @PublicEvolving
   def assignAscendingTimestamps(extractor: T => Long): DataStream[T] = {
     val cleanExtractor = clean(extractor)
     val extractorFunction = new AscendingTimestampExtractor[T] {
@@ -714,7 +715,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * written.
    *
    */
-  @Experimental
+  @PublicEvolving
   def print(): DataStreamSink[T] = stream.print()
 
   /**
@@ -725,7 +726,7 @@ class DataStream[T](stream: JavaStream[T]) {
    *
    * @return The closed DataStream.
    */
-  @Experimental
+  @PublicEvolving
   def printToErr() = stream.printToErr()
 
   /**
@@ -735,7 +736,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param path The path pointing to the location the text file is written to
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsText(path: String): DataStreamSink[T] =
     stream.writeAsText(path, 0L)
 
@@ -749,7 +750,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * @param millis The file update frequency
    * @return The closed DataStream
    */
-  @Experimental
+  @PublicEvolving
   def writeAsText(path: String, millis: Long): DataStreamSink[T] =
     stream.writeAsText(path, millis)
 
@@ -762,7 +763,7 @@ class DataStream[T](stream: JavaStream[T]) {
     *                  OVERWRITE.
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsText(path: String, writeMode: FileSystem.WriteMode): DataStreamSink[T] = {
     if (writeMode != null) {
       stream.writeAsText(path, writeMode)
@@ -782,7 +783,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param millis The file update frequency
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsText(
       path: String,
       writeMode: FileSystem.WriteMode,
@@ -802,7 +803,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param path Path to the location of the CSV file
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsCsv(path: String): DataStreamSink[T] = {
     writeAsCsv(
       path,
@@ -820,7 +821,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param millis File update frequency
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsCsv(path: String, millis: Long): DataStreamSink[T] = {
     writeAsCsv(
       path,
@@ -838,7 +839,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param writeMode Controls whether an existing file is overwritten or not
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsCsv(path: String, writeMode: FileSystem.WriteMode): DataStreamSink[T] = {
     writeAsCsv(
       path,
@@ -857,7 +858,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param millis File update frequency
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsCsv(path: String, writeMode: FileSystem.WriteMode, millis: Long): DataStreamSink[T]
= {
     writeAsCsv(
       path,
@@ -878,7 +879,7 @@ class DataStream[T](stream: JavaStream[T]) {
     * @param fieldDelimiter Delimiter for consecutive fields
     * @return The closed DataStream
     */
-  @Experimental
+  @PublicEvolving
   def writeAsCsv(
       path: String,
       writeMode: FileSystem.WriteMode,
@@ -898,7 +899,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * Writes a DataStream using the given [[OutputFormat]]. The
    * writing is performed periodically, in every millis milliseconds.
    */
-  @Experimental
+  @PublicEvolving
   def write(format: OutputFormat[T], millis: Long): DataStreamSink[T] = {
     stream.write(format, millis)
   }
@@ -907,7 +908,7 @@ class DataStream[T](stream: JavaStream[T]) {
    * Writes the DataStream to a socket as a byte array. The format of the output is
    * specified by a [[SerializationSchema]].
    */
-  @Experimental
+  @PublicEvolving
   def writeToSocket(
       hostname: String,
       port: Integer,

http://git-wip-us.apache.org/repos/asf/flink/blob/572855da/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/JoinedStreams.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/JoinedStreams.scala
b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/JoinedStreams.scala
index b6fbadf..21c5d84 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/JoinedStreams.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/JoinedStreams.scala
@@ -18,7 +18,7 @@
 
 package org.apache.flink.streaming.api.scala
 
-import org.apache.flink.annotation.{Experimental, Public}
+import org.apache.flink.annotation.{PublicEvolving, Public}
 import org.apache.flink.api.common.functions.{FlatJoinFunction, JoinFunction}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.functions.KeySelector
@@ -148,7 +148,7 @@ object JoinedStreams {
     /**
      * Specifies the window on which the join operation works.
      */
-    @Experimental
+    @PublicEvolving
     def window[W <: Window](
         assigner: WindowAssigner[_ >: JavaCoGroupedStreams.TaggedUnion[T1, T2], W])
         : JoinedStreams.WithWindow[T1, T2, KEY, W] = {
@@ -197,7 +197,7 @@ object JoinedStreams {
     /**
      * Sets the [[Trigger]] that should be used to trigger window emission.
      */
-    @Experimental
+    @PublicEvolving
     def trigger(newTrigger: Trigger[_ >: JavaCoGroupedStreams.TaggedUnion[T1, T2], _ >:
W])
     : JoinedStreams.WithWindow[T1, T2, KEY, W] = {
       new WithWindow[T1, T2, KEY, W](
@@ -216,7 +216,7 @@ object JoinedStreams {
      * Note: When using an evictor window performance will degrade significantly, since
      * pre-aggregation of window results cannot be used.
      */
-    @Experimental
+    @PublicEvolving
     def evictor(newEvictor: Evictor[_ >: JavaCoGroupedStreams.TaggedUnion[T1, T2], _ >:
W])
     : JoinedStreams.WithWindow[T1, T2, KEY, W] = {
       new WithWindow[T1, T2, KEY, W](

http://git-wip-us.apache.org/repos/asf/flink/blob/572855da/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
index 271796b..923aad6 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/KeyedStream.scala
@@ -18,7 +18,7 @@
 
 package org.apache.flink.streaming.api.scala
 
-import org.apache.flink.annotation.{Experimental, Internal, Public}
+import org.apache.flink.annotation.{PublicEvolving, Internal, Public}
 import org.apache.flink.api.common.functions._
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.common.typeutils.TypeSerializer
@@ -112,7 +112,7 @@ class KeyedStream[T, K](javaStream: KeyedJavaStream[T, K]) extends DataStream[T]
    * @param assigner The `WindowAssigner` that assigns elements to windows.
    * @return The trigger windows data stream.
    */
-  @Experimental
+  @PublicEvolving
   def window[W <: Window](assigner: WindowAssigner[_ >: T, W]): WindowedStream[T, K,
W] = {
     new WindowedStream(new WindowedJavaStream[T, K, W](javaStream, assigner))
   }

http://git-wip-us.apache.org/repos/asf/flink/blob/572855da/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
index 4317931..58b100e 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/StreamExecutionEnvironment.scala
@@ -19,7 +19,7 @@
 package org.apache.flink.streaming.api.scala
 
 import com.esotericsoftware.kryo.Serializer
-import org.apache.flink.annotation.{Internal, Experimental, Public}
+import org.apache.flink.annotation.{Internal, PublicEvolving, Public}
 import org.apache.flink.api.common.io.{FileInputFormat, InputFormat}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.api.java.typeutils.runtime.kryo.KryoSerializer
@@ -91,7 +91,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * avoiding serialization and de-serialization.
    *
    */
-  @Experimental
+  @PublicEvolving
   def disableOperatorChaining(): StreamExecutionEnvironment = {
     javaEnv.disableOperatorChaining()
     this
@@ -127,7 +127,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    *           If true checkpointing will be enabled for iterative jobs as well.
    */
   @deprecated
-  @Experimental
+  @PublicEvolving
   def enableCheckpointing(interval : Long,
                           mode: CheckpointingMode,
                           force: Boolean) : StreamExecutionEnvironment = {
@@ -191,7 +191,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * failure the job will be resubmitted to the cluster indefinitely.
    */
   @deprecated
-  @Experimental
+  @PublicEvolving
   def enableCheckpointing() : StreamExecutionEnvironment = {
     javaEnv.enableCheckpointing()
     this
@@ -218,7 +218,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * program can be executed highly available and strongly consistent (assuming that Flink
    * is run in high-availability mode).
    */
-  @Experimental
+  @PublicEvolving
   def setStateBackend(backend: AbstractStateBackend): StreamExecutionEnvironment = {
     javaEnv.setStateBackend(backend)
     this
@@ -227,7 +227,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
   /**
    * Returns the state backend that defines how to store and checkpoint state.
    */
-  @Experimental
+  @PublicEvolving
   def getStateBackend: AbstractStateBackend = javaEnv.getStateBackend()
   
   /**
@@ -235,7 +235,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * effectively disables fault tolerance. A value of "-1" indicates that the system
    * default value (as defined in the configuration) should be used.
    */
-  @Experimental
+  @PublicEvolving
   def setNumberOfExecutionRetries(numRetries: Int): Unit = {
     javaEnv.setNumberOfExecutionRetries(numRetries)
   }
@@ -245,7 +245,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * of "-1" indicates that the system default value (as defined in the configuration)
    * should be used.
    */
-  @Experimental
+  @PublicEvolving
   def getNumberOfExecutionRetries = javaEnv.getNumberOfExecutionRetries
 
   // --------------------------------------------------------------------------------------------
@@ -327,7 +327,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    *
    * @param characteristic The time characteristic.
    */
-  @Experimental
+  @PublicEvolving
   def setStreamTimeCharacteristic(characteristic: TimeCharacteristic) : Unit = {
     javaEnv.setStreamTimeCharacteristic(characteristic)
   }
@@ -336,10 +336,9 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * Gets the time characteristic/
    *
    * @see #setStreamTimeCharacteristic
-   *
    * @return The time characteristic.
    */
-  @Experimental
+  @PublicEvolving
   def getStreamTimeCharacteristic = javaEnv.getStreamTimeCharacteristic()
 
   // --------------------------------------------------------------------------------------------
@@ -474,7 +473,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * character set. The maximum retry interval is specified in seconds, in case
    * of temporary service outage reconnection is initiated every second.
    */
-  @Experimental
+  @PublicEvolving
   def socketTextStream(hostname: String, port: Int, delimiter: Char = '\n', maxRetry: Long
= 0):
     DataStream[String] =
     javaEnv.socketTextStream(hostname, port)
@@ -485,7 +484,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
    * determine the type of the data produced by the input format. It will attempt to determine
the
    * data type by reflection, unless the input format implements the ResultTypeQueryable
interface.
    */
-  @Experimental
+  @PublicEvolving
   def createInput[T: ClassTag : TypeInformation](inputFormat: InputFormat[T, _]): DataStream[T]
=
     javaEnv.createInput(inputFormat)
 
@@ -562,6 +561,7 @@ class StreamExecutionEnvironment(javaEnv: JavaEnv) {
 
   /**
    * Getter of the wrapped [[org.apache.flink.streaming.api.environment.StreamExecutionEnvironment]]
+ *
    * @return The encased ExecutionEnvironment
    */
   @Internal
@@ -590,7 +590,7 @@ object StreamExecutionEnvironment {
    * @param parallelism
    * The parallelism to use as the default local parallelism.
    */
-  @Experimental
+  @PublicEvolving
   def setDefaultLocalParallelism(parallelism: Int) : Unit =
     StreamExecutionEnvironment.setDefaultLocalParallelism(parallelism)
 

http://git-wip-us.apache.org/repos/asf/flink/blob/572855da/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
----------------------------------------------------------------------
diff --git a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
index 15b9505..6385831 100644
--- a/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
+++ b/flink-streaming-scala/src/main/scala/org/apache/flink/streaming/api/scala/WindowedStream.scala
@@ -18,7 +18,7 @@
 
 package org.apache.flink.streaming.api.scala
 
-import org.apache.flink.annotation.{Experimental, Public}
+import org.apache.flink.annotation.{PublicEvolving, Public}
 import org.apache.flink.api.common.functions.{FoldFunction, ReduceFunction}
 import org.apache.flink.api.common.typeinfo.TypeInformation
 import org.apache.flink.streaming.api.datastream.{WindowedStream => JavaWStream}
@@ -64,7 +64,7 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K,
W]) {
   /**
    * Sets the [[Trigger]] that should be used to trigger window emission.
    */
-  @Experimental
+  @PublicEvolving
   def trigger(trigger: Trigger[_ >: T, _ >: W]): WindowedStream[T, K, W] = {
     javaStream.trigger(trigger)
     this
@@ -76,7 +76,7 @@ class WindowedStream[T, K, W <: Window](javaStream: JavaWStream[T, K,
W]) {
    * Note: When using an evictor window performance will degrade significantly, since
    * pre-aggregation of window results cannot be used.
    */
-  @Experimental
+  @PublicEvolving
   def evictor(evictor: Evictor[_ >: T, _ >: W]): WindowedStream[T, K, W] = {
     javaStream.evictor(evictor)
     this


Mime
View raw message