shuiqiangchen commented on a change in pull request #13803:
URL: https://github.com/apache/flink/pull/13803#discussion_r514034737
##########
File path: flink-python/pyflink/datastream/functions.py
##########
@@ -540,3 +540,155 @@ def __init__(self, sink_func: Union[str, JavaObject]):
:param sink_func: The java SinkFunction object or the full name of the
SinkFunction class.
"""
super(SinkFunction, self).__init__(sink_func)
+
+
+class ProcessFunction(Function):
+ """
+ A function that process elements of a stream.
+
+ For every element in the input stream process_element(value, ctx, out) is
invoked. This can
+ produce zero or more elements as output. Implementations can also query
the time and set timers
+ through the provided Context. For firing timers on_timer(long, ctx, out)
will be invoked. This
+ can again produce zero or more elements as output and register further
timers.
+
+ Note that access to keyed state and timers (which are also scoped to a
key) is only available if
+ the ProcessFunction is applied on a KeyedStream. Process Function is
always a RichFunction.
+ Therefore, access to the RuntimeContext is always available and setup and
teardown methods can
+ be implemented. See Function.open() and Function.close().
+ """
+
+ @abc.abstractmethod
+ def process_element(self, value, ctx: 'Context', out: 'Collector'):
+ """
+ Process one element from the input stream.
+
+ This function can output zero or more elements using the Collector
parameter and also update
+ internal state or set timers using the Context parameter.
+
+ :param value: The input value.
+ :param ctx: A Context that allows querying the timestamp of the
element and getting a
+ TimerService for registering timers and querying the
time. The context is only
+ valid during the invocation of this method, do not store
it.
+ :param out: The collector for returning result values.
+ """
+ pass
+
+ @abc.abstractmethod
+ def on_timer(self, timestamp, ctx: 'OnTimerContext', out: 'Collector'):
+ """
+ Called when a timer set using TimerService fires.
+
+ :param timestamp: The timestamp of the firing timer.
+ :param ctx: An OnTimerContext that allows querying the timestamp of
the firing timer,
+ querying the TimeDomain of the firing timer and getting a
TimerService for
+ registering timers and querying the time. The context is
only valid during the
+ invocation of this method, do not store it.
+ :param out: The collector for returning result values.
+ """
+ pass
+
+ class Context(abc.ABC):
+ """
+ Information available in an invocation of process_element(value, ctx,
out) or
+ on_timer(value, ctx, out).
+ """
+
+ @abc.abstractmethod
+ def timestamp(self):
+ """
+ Timestamp of the element currently being processed or timestamp of
a firing timer.
+
+ This might be null, for example if the time characteristic of your
program is set to
+ TimeCharacteristic.ProcessingTime.
+ """
+ pass
+
+ @abc.abstractmethod
+ def timer_service(self):
+ """
+ A Timer service for querying time and registering timers.
+ """
+ pass
+
+ class OnTimerContext(abc.ABC):
+ """
+ Information available in an invocation of on_timer(long,
OnTimerContext, Collector)
+ """
+ @abc.abstractmethod
+ def time_domain(self):
Review comment:
Maybe we can remove this interface currently
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
[email protected]
