[FLINK-4460] Allow ProcessFunction on non-keyed streams #3438
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -19,30 +19,37 @@ | |
| package org.apache.flink.streaming.api.functions; | ||
|
|
||
| import org.apache.flink.annotation.PublicEvolving; | ||
| import org.apache.flink.api.common.functions.AbstractRichFunction; | ||
| import org.apache.flink.streaming.api.TimeDomain; | ||
| import org.apache.flink.streaming.api.TimerService; | ||
| import org.apache.flink.util.Collector; | ||
|
|
||
| /** | ||
| * A function that processes elements of a stream. | ||
| * | ||
| * <p>For every element in the input stream {@link #processElement(Object, Context, Collector)} | ||
| * is invoked. This can produce zero or more elements as output. Implementations can also | ||
| * query the time and set timers through the provided {@link Context}. For firing timers | ||
| * {@link #onTimer(long, OnTimerContext, Collector)} will be invoked. This can again produce | ||
| * zero or more elements as output and register further timers. | ||
| * | ||
| * <p><b>NOTE:</b> Access to keyed state and timers (which are also scoped to a key) is only | ||
| * available if the {@code ProcessFunction} is applied on a {@code KeyedStream}. | ||
| * | ||
| * <p><b>NOTE:</b> A {@code ProcessFunction} is always a | ||
| * {@link org.apache.flink.api.common.functions.RichFunction}. Therefore, access to the | ||
| * {@link org.apache.flink.api.common.functions.RuntimeContext} is always available and setup and | ||
| * teardown methods can be implemented. See | ||
| * {@link org.apache.flink.api.common.functions.RichFunction#open(org.apache.flink.configuration.Configuration)} | ||
| * and {@link org.apache.flink.api.common.functions.RichFunction#close()}. | ||
| * | ||
| * @param <I> Type of the input elements. | ||
| * @param <O> Type of the output elements. | ||
| */ | ||
| @PublicEvolving | ||
| public abstract class ProcessFunction<I, O> extends AbstractRichFunction { | ||
|
There was a problem hiding this comment. hi, changing form There was a problem hiding this comment. I think the problem is that we need a default implementation for There was a problem hiding this comment. hi @wenlong88 in the ML discussion (https://lists.apache.org/thread.html/f3fe7d68986877994ad6b66173f40e72fc454420720a74ea5a834cc2@%3Cdev.flink.apache.org%3E) we decided to make |
||
|
|
||
| private static final long serialVersionUID = 1L; | ||
|
|
||
| /** | ||
| * Process one element from the input stream. | ||
|
|
@@ -59,7 +66,7 @@ public interface ProcessFunction<I, O> extends Function { | |
| * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation | ||
| * to fail and may trigger recovery. | ||
| */ | ||
| public abstract void processElement(I value, Context ctx, Collector<O> out) throws Exception; | ||
|
|
||
| /** | ||
| * Called when a timer set using {@link TimerService} fires. | ||
|
|
@@ -74,36 +81,36 @@ public interface ProcessFunction<I, O> extends Function { | |
| * @throws Exception This method may throw exceptions. Throwing an exception will cause the operation | ||
| * to fail and may trigger recovery. | ||
| */ | ||
| public void onTimer(long timestamp, OnTimerContext ctx, Collector<O> out) throws Exception {} | ||
|
|
||
| /** | ||
| * Information available in an invocation of {@link #processElement(Object, Context, Collector)} | ||
| * or {@link #onTimer(long, OnTimerContext, Collector)}. | ||
| */ | ||
| public abstract class Context { | ||
|
|
||
| /** | ||
| * Timestamp of the element currently being processed or timestamp of a firing timer. | ||
| * | ||
| * <p>This might be {@code null}, for example if the time characteristic of your program | ||
| * is set to {@link org.apache.flink.streaming.api.TimeCharacteristic#ProcessingTime}. | ||
| */ | ||
| public abstract Long timestamp(); | ||
|
|
||
| /** | ||
| * A {@link TimerService} for querying time and registering timers. | ||
| */ | ||
| public abstract TimerService timerService(); | ||
| } | ||
|
|
||
| /** | ||
| * Information available in an invocation of {@link #onTimer(long, OnTimerContext, Collector)}. | ||
| */ | ||
| public abstract class OnTimerContext extends Context { | ||
| /** | ||
| * The {@link TimeDomain} of the firing timer. | ||
| */ | ||
| public abstract TimeDomain timeDomain(); | ||
| } | ||
|
|
||
| } | ||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is this internal method only exposed as
publicfor the Scala API? If yes, I'm wondering if it makes sense to calltransformmanually in the ScalaDataStreamAPI.There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Yes, it's exposed for that. The pattern, so far, is for methods to also expose a public method that takes a
TypeInformationbecause we get theTypeInformationfrom the context bound in the Scala API.Calling
transform()manually is an option but if we do that we would basically not base the Scala API on the Java API anymore and we would have code that instantiates the Stream Operators in both the Java and Scala API. For example, right now we have the code for instantiating a flat map operator in(Java)DataStreamwhile(Scala)DataStream.flatMap()calls that method.What do you think?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Makes sense to keep it like that. The benefits to base the Scala API on top of the Java API instead of duplicating it are very persuasive, too. 😄 So +1 to keep it as is. 👍 I was just wondering whether users would be confused by this.