Hi team,
I have one question, hoping to get some help.
Will BoundedOutOfOrderness have any impact on the KeyedProcessFunction? If so,
in what way it can impact KeyedProcessFunction?
Thanks!
Best,
Lijuan
veStreams.keyBy()
("...Apply the partitioning on the input and feedback streams instead")
does not look right.
I tried that (I made a loop with a single stream of
Either) but it seems there is no way
of processing an IterativeStream with a KeyedProcessFunction, nor to feed
back a K
I am trying to implement an iterative streaming job that processes the loop
> with a KeyedProcessFunction.
>
> I need a KeyedProcessFunction to use keyed state and to emit a side-output
> (that after further transformations becomes the feedback)
>
> Problem is Iterativ
Hi all,
I am trying to implement an iterative streaming job that processes the loop
with a KeyedProcessFunction.
I need a KeyedProcessFunction to use keyed state and to emit a side-output
(that after further transformations becomes the feedback)
Problem is IterativeStream.process() only accepts
Yes, restarting the app with a clean state does seem to fix the issue, but
I think I may have found a bug in Flink.
Here's how we can replicate it:
- Create a simple application with KeyedProcessFunction (with onTimer())
- Send a few records with the same key. In processElement(), register
After fixing your negative timestamp bug, can the timer be triggered?
> On 23 Mar 2022, at 2:39 AM, Binil Benjamin wrote:
>
> Here are some more findings as I was debugging this. I peeked into the
> snapshot to see the current values in "_timer_state/processing_user-timers"
> and here is
Here are some more findings as I was debugging this. I peeked into the
snapshot to see the current values in
"_timer_state/processing_user-timers" and here is how they look:
Timer{timestamp=-9223372036854715808, key=(FFX22...),
namespace=VoidNamespace}
Timer{timestamp=-9223372036854715808,
Hi,
Parallelism is currently set to 9 and it appears to be occurring for all
subtasks.
We did put logs to see the various timestamps. The following logs are from
the last 5 days.
- logs from processElement() - logged immediately after timer registration:
"message":
e: onTimer() of a KeyedProcessFunction stops getting triggered after a
while
Hi,
Unfortunately, I cannot share the entire code, but the class roughly looks like
this:
public class WfProcessFunction extends KeyedProcessFunction, Map, Map> {
@Override
public void processElement(Ma
Hi,
Unfortunately, I cannot share the entire code, but the class roughly looks
like this:
public class WfProcessFunction extends KeyedProcessFunction, Map, Map> {
@Override
public void processElement(Map inputRecord,
Context context, Collector> collector) throws
Exc
Hi, can you share your code so we can check whether it is written correctly.
> On 18 Mar 2022, at 7:54 AM, Binil Benjamin wrote:
>
> Hi,
>
> We have a class that extends KeyedProcessFunction and overrides onTimer()
> method. During processElement(), we register a ti
Hi,
We have a class that extends KeyedProcessFunction and overrides onTimer()
method. During processElement(), we register a timer callback using
context.timerService().registerProcessingTimeTimer(). For
a while, we see that the onTimer() method is getting called back and
everything works
??KeyedProcessFunctionprocessElementKeyBy??processElement100
= aggrRecord.get(); // Always get null value.
Thanks,
Suchithra
From: JING ZHANG
Sent: Wednesday, June 9, 2021 2:20 PM
To: V N, Suchithra (Nokia - IN/Bangalore)
Cc: user@flink.apache.org; Jash, Shaswata (Nokia - IN/Bangalore)
Subject: Re: Issue with onTimer method of KeyedProcessFunction
Hi Suchithra
= aggrRecord.get(); // Always get null value.
Thanks,
Suchithra
From: JING ZHANG
Sent: Wednesday, June 9, 2021 2:20 PM
To: V N, Suchithra (Nokia - IN/Bangalore)
Cc: user@flink.apache.org; Jash, Shaswata (Nokia - IN/Bangalore)
Subject: Re: Issue with onTimer method of KeyedProcessFunction
Hi Suchithra
n. After upgrading to 1.12.3 version, the onTimer method of
> KeyedProcessFunction is not behaving correctly, the value of ReducingState
> and ValueState always return null.
>
>
>
> Could you please help in debugging the issue.
>
>
>
> Thanks,
>
> Suchithra
>
>
>
Hello,
We are using apache flink 1.7 and now trying to upgrade to flink 1.12.3
version. After upgrading to 1.12.3 version, the onTimer method of
KeyedProcessFunction is not behaving correctly, the value of ReducingState and
ValueState always return null.
Could you please help in debugging
ot be a problem?
>
> Thanks, once again.
>
> Dawid Wysakowicz escreveu no dia segunda,
> 10/05/2021 à(s) 09:13:
>
>> Hey Miguel,
>>
>> I think you could take a look at the CepOperator which does pretty much
>> what you are describing.
>>
>
t answers for your questions. If you use
> KeyedProcessFunction it is always scoped to a single Key. There is no way
> to process events from other keys. If you want to have more control over
> state and e.g. use PriorityQueue which would be snapshotted on checkpoint
> you could look
Hey Miguel,
I think you could take a look at the CepOperator which does pretty much
what you are describing.
As for more direct answers for your questions. If you use
KeyedProcessFunction it is always scoped to a single Key. There is no
way to process events from other keys. If you want to have
Hi Timo,
Thanks for your answer. I think I wasn't clear enough in my initial
message, so let me give more details.
The stream is not keyed by timestamp, it's keyed by a custom field (e.g.,
user-id) and then fed into a KeyedProcessFunction. I want to process all
events for a given user in order
, Miguel Araújo wrote:
Hi everyone,
I have a KeyedProcessFunction whose events I would like to process in
event-time order.
My initial idea was to use a Map keyed by timestamp and, when a new
event arrives, iterate over the Map to process events older than the
current watermark.
The issue
Hi everyone,
I have a KeyedProcessFunction whose events I would like to process in
event-time order.
My initial idea was to use a Map keyed by timestamp and, when a new event
arrives, iterate over the Map to process events older than the current
watermark.
The issue is that I obviously can't use
I missed in documentation:
A KeyedProcessFunction is always a RichFunction. Therefore, access to the
RuntimeContext is always available and setup and teardown methods can be
implemented. See
RichFunction.open(org.apache.flink.configuration.Configuration) and
RichFunction.close().
https
Hello,
I'm learning State Processor API:
https://ci.apache.org/projects/flink/flink-docs-stable/dev/libs/state_processor_api.html
There is example in this page with StatefulFunctionWithTime extends
KeyedProcessFunction. And here we can see method open() we need implement to
initialize state
below.
>>>
>>> I managed to figure out the root cause is that Flink stream
>>> execution environment has a default parallelism as 8.*I
didn't
>>> notice in the doc, could the Community add this
explicitly into
>>> I have sent another email with the title 'BoundedOutOfOrderness
> >>> Watermark Generator is NOT making the event time to advance'
> >>> using another email of mine, fuyao...@oracle.com
> >>> <mailto:fuyao...@oracle.com
reflected in the Flink Cluster web UI interface. I can see the
watermark is advancing, but it is not in reality. *That's
causing the inconsistency problem I mentioned in the other email
I mentioned above. Will this be considered as a bug in the UI?*
My current ques
e doc, could the Community add this explicitly into
> the official doc to avoid some confusion? Thanks.*
>
> From my understanding, the watermark advances based on the
> lowest watermark among the 8, so I can not advance the bound out
> of orderness watermark since I
lity. *That's
causing the inconsistency problem I mentioned in the other
email
I mentioned above. Will this be considered as a bug in the UI?*
My current question is, since I have full outer join operation
before the KeyedProcessFunction here. How can I let the
bound o
ark is advancing, but it is not in reality. *That's
causing the inconsistency problem I mentioned in the other email
I mentioned above. Will this be considered as a bug in the UI?*
My current question is, since I have full outer join operation
before the KeyedProcessFu
ster web UI interface. I can see the
watermark is advancing, but it is not in reality. *That's
causing the inconsistency problem I mentioned in the other email
I mentioned above. Will this be considered as a bug in the UI?*
My current question is, since I have full out
of parallelism 1, it will reflect
>> the watermark in the context correctly. One more thing is that this
>> behavior is not reflected in the Flink Cluster web UI interface. I can see
>> the watermark is advancing, but it is not in reality. *That's causing
>> the inconsistency problem I me
s advancing, but it is not in reality. *That's causing the
> inconsistency problem I mentioned in the other email I mentioned above.
> Will this be considered as a bug in the UI?*
>
> My current question is, since I have full outer join operation before the
> KeyedProcessF
her email I mentioned above.
Will this be considered as a bug in the UI?*
My current question is, since I have full outer join operation before the
KeyedProcessFunction here. How can I let the bound of orderness watermark /
punctuated watermark strategy work if the parallelism > 1? It can only
upda
r 10 seconds.
>>
>> In the first stage, I first consume three Kafka streams and transform it
>> to Flink Datastream using a deserialization schema containing some type and
>> date format transformation, and then I register these data streams as Table
>> and do a full outer join on
some type and
> date format transformation, and then I register these data streams as Table
> and do a full outer join one by one using Table API. I also add query
> configuration for this to avoid excessive state. The primary key is also
> the join key.
>
> In the second sta
the joined table to a retracted stream and
put it into KeyedProcessFunction to generate the business object if the
business object's primary key is inactive for 10 second.
Is this way of handling the data the suggested approach? (I understand I
can directly consume kafka data in Table API. I haven't
Trenikhun
Cc: Dawid Wysakowicz ; Flink User Mail List
Subject: Re: Unit Test for KeyedProcessFunction with out-of-core state
The new backend would be for unit tests (instead of a RocksDB mock). It's kind
of the mock for out-of-core behavior that you initially requested.
To use rocksDB in an IT Case
-
> *From:* Arvid Heise
> *Sent:* Monday, September 14, 2020 4:26:47 AM
> *To:* Dawid Wysakowicz
> *Cc:* Alexey Trenikhun ; Flink User Mail List <
> user@flink.apache.org>
> *Subject:* Re: Unit Test for KeyedProcessFunction with out-of-core state
>
> Hi Ale
for KeyedProcessFunction with out-of-core state
Hi Alexey,
Definition of test levels are always a bit blurry when writing tests for a data
processing framework, but I'm convinced that in your case, you should rather
think in terms of integration tests than unit tests:
* Unit test should really just
; Alexey
>
> --
> *From:* Tzu-Li (Gordon) Tai
> *Sent:* Friday, September 4, 2020 12:35:48 AM
> *To:* Alexey Trenikhun
> *Cc:* Flink User Mail List
> *Subject:* Re: Unit Test for KeyedProcessFunction with out-of-core state
>
> Hi Alexey,
>
, September 4, 2020 12:35:48 AM
To: Alexey Trenikhun
Cc: Flink User Mail List
Subject: Re: Unit Test for KeyedProcessFunction with out-of-core state
Hi Alexey,
Is there a specific reason why you want to test against RocksDB?
Otherwise, in Flink tests we use a `KeyedOneInputStreamOperatorTestHarness
/1d5f44710270d1c615537f0d05ab49e699d3a6e5/flink-streaming-java/src/test/java/org/apache/flink/streaming/util/KeyedOneInputStreamOperatorTestHarness.java
On Fri, Sep 4, 2020 at 2:28 AM Alexey Trenikhun wrote:
> Hello,
> I want to unit test KeyedProcessFunction which uses with out-of-core state
> (like rocksdb).
> Do
Hello,
I want to unit test KeyedProcessFunction which uses with out-of-core state
(like rocksdb).
Does Flink has mock for rocksdb, which can be used in unit tests ?
Thanks,
Alexey
ctx.timestamp()其实就是获取的StreamRecord的时间戳，也就是事件被提取出来的时间戳。
这个方法一般需要使用event time，并且在数据流上assign过timestamp和watermark。
ゞ野蠻遊戲χ 于2020年8月16日周日 下午7:57写道：
> 大家好
>
>
> 当我在KeyedProcessFunction的processElement方法中获取processingTime，就像这样ctx.timestamp()，返回为null，我改如何在processElement中获取processingTime？
>
>
> 谢谢！
> 嘉治
根据Context获取timerService，然后获取处理时间即可。
在 2020/8/16 下午7:57，“ゞ野蠻遊戲χ” 写入:
大家好
当我在KeyedProcessFunction的processElement方法中获取processingTime，就像这样ctx.timestamp()，返回为null，我改如何在processElement中获取processingTime？
谢谢！
嘉治
e event time with an AscendingTimestampExtractor
>
> I noticed when debugging that in the KeyedProcessFunction that
> after my highest known event time of: 2020-06-23T00:46:30.000Z
>
> the processElement method had a watermark with an impossible date of:
> -292275055-05-16T16:47:
My source is a Kafka topic.
I am using Event Time.
I assign the event time with an AscendingTimestampExtractor
I noticed when debugging that in the KeyedProcessFunction that
after my highest known event time of: 2020-06-23T00:46:30.000Z
the processElement method had a watermark
Hi
+1. Because there is no need to generate an instance for each key, flink just
maintain the key collection in one instance. Imagine what would happen if the
number of keys were unlimited.
Best,
Yichao Yang
--Original--
From:"Tzu-Li (Gordon)
Hi,
Records with the same key will be processed by the same partition.
Note there isn't an instance of a keyed process function for each key.
There is a single instance per partition, and all keys that are distributed
to the same partition will get processed by the same keyed process function
Hi All,
I have a keyed data stream and calling a keyedProcessFunction after keyBy
operation on datastream. Till now my understanding was, "For all different n-
elements in keyed stream if their keys are same, same instance of
keyedProcessFunction is called and for another ele
is constantly emitting the data and
>> bumping your timers? Keep in mind that the code that you are basing on has
>> the following characteristic:
>>
>> > In the following example a KeyedProcessFunction maintains counts per
>> key, and emits a key/count pair whe
com>*
On Tue, Jun 18, 2019 at 9:22 AM Piotr Nowojski wrote:
> Hi,
>
> Isn’t your problem that the source is constantly emitting the data and
> bumping your timers? Keep in mind that the code that you are basing on has
> the following characteristic:
>
> > In the following
Hi,
Isn’t your problem that the source is constantly emitting the data and bumping
your timers? Keep in mind that the code that you are basing on has the
following characteristic:
> In the following example a KeyedProcessFunction maintains counts per key, and
> emits a key/count pair wh
Hi,
I used this example of KeyedProcessFunction from the FLink website [1] and
I have implemented my own KeyedProcessFunction to process some
approximation counting [2]. This worked very well. Then I switched the data
source to consume strings from Twitter [3]. The data source is consuming
Thanks Bowen.
On Thu, Jul 19, 2018 at 4:45 PM, Bowen Li wrote:
> Hi Anna,
>
> KeyedProcessFunction is only available starting from Flink 1.5. The doc is
> here
> <https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html#the-ke
Hi Anna,
KeyedProcessFunction is only available starting from Flink 1.5. The doc is
here
<https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/operators/process_function.html#the-keyedprocessfunction>.
It extends ProcessFunction and shares the same functionalities except
givin
Hello all,
I am using Flink 1.4 because thats the version provided by the latest AWS
EMR.
Is KeyedProcessFunction available in Flink 1.4?
Also please share any links to good examples on using KeyedProcessFunction
.
Thanks
59 matches
Mail list logo
