Unsubscribe

[Dan Pettersson]

Unsubscribe

Re: Unsubscribe

[Dan Pettersson]

I've also tried a few times now the last couple of months. I think it would
be very nice if the "flink admin" could look into this, instead of us
reaching out to the Apache Infrastructure team.

Thanks,

/Dan

Den tors 6 maj 2021 kl 13:31 skrev Chesnay Schepler :

> Could you reach out to the Apache Infrastructure team
>  about not being able to
> unsubscribe? Maybe this functionality is currently broken.
>
> On 5/6/2021 12:48 PM, Andrew Kramer wrote:
>
> I have been unable to unsubscribe as well. Have tried emailing just like
> you
>
> On Thu, May 6, 2021 at 3:33 AM Xander Song  wrote:
>
>> How can I unsubscribe from the Apache Flink user mailing list? I have
>> tried emailing user-unsubscr...@flink.apache.org, but am still receiving
>> messages.
>>
>> Thank you.
>>
>
>

Re: A couple of question for Stateful Functions

[Dan Pettersson]

Thanks for your quick reply.

/Dan

Den ons 2 sep. 2020 kl 12:24 skrev Igal Shilman :

> Hi Dan, let me try to answer your questions:
>
>
>> I guess my question is if one can
>> freely mix Flink core with SF's code with regards to performance,
>> fault-tolerance, and checkpointing?
>
>
> The main limitations at the moment is that, currently SF requires a
> processing time watermark semantics only, event time is not supported as it
> is difficult to reason about completion in the presence of loops.
> Other than that in respect to fault-tolerance and checkpointing StateFun
> is built on top of the DataStream API, so the same gurantines applies to SF
> as-well.
>
>
>> I'm unable to use Protobuf so POJO's is going to be processed. Is there a
>> large performance impact of using Pojos instead of protos in SF?
>
>
> The only supported message data type for SF is Protocol Buffers and I
> would highly recommend to use that,
> One option is to transform your Pojo into a Protobuf just before entering
> SF, and you can convert it back to your original Pojo when exiting from SF.
> If you absolutely have to use something else, you can fall back to kryo[1]
> or provide your own[2] but then schema evaluation of your messages and
> state is not guaranteed anymore, and you
> lose the ability to communicate with remote functions.
>
> Would the appendingBuffer be
>> de-/serialized for each function invocation?
>
>
> The appending buffer supports efficient appends (the buffer is *not*
> deserialized on every function invocation)
> In fact, it is backed by Flink's ListState[3]
>
> [1] -
> https://github.com/apache/flink-statefun/blob/master/statefun-examples/statefun-flink-datastream-example/src/main/java/org/apache/flink/statefun/examples/datastream/Example.java#L62
> [2] -
> https://github.com/apache/flink-statefun/blob/master/statefun-flink/statefun-flink-core/src/main/java/org/apache/flink/statefun/flink/core/message/MessagePayloadSerializer.java
> [3] -
> https://ci.apache.org/projects/flink/flink-docs-stable/dev/stream/state/state.html#using-keyed-state
>
> Thanks,
> Igal
>
> On Wed, Sep 2, 2020 at 10:51 AM Till Rohrmann 
> wrote:
>
>> Hi Dan,
>>
>> thanks for reaching out to the Flink community. I'm pulling in Gordon and
>> Igal who will be able to answer your questions.
>>
>> Cheers,
>> Till
>>
>> On Wed, Sep 2, 2020 at 8:22 AM danp  wrote:
>>
>>> Hi,
>>>
>>> Nice to see the progress of Stateful functions.
>>>
>>> I have a few questions that I hope you can reply to.
>>>
>>> My first question is regarding the newly implemented
>>> StatefulFunctionDataStreamBuilder.
>>> Is there anything to pay attention to if one first union a couple of
>>> streams
>>> and performs a sort via a keyBy and a KeyedProcessFunction before
>>> dispatching the messages to via RoutableMessage?
>>> In this sorting I'm using a mapstate and I guess my question is if one
>>> can
>>> freely mix Flink core with SF's code with regards to performance,
>>> fault-tolerance, and checkpointing?
>>>
>>> I'm unable to use Protobuf so POJO's is going to be processed. Is there a
>>> large performance impact of using Pojos instead of protos in SF?
>>>
>>> Also I wonder if there's going to be a severe performance penalty if I
>>> had a
>>> function that was called very often lets say 1000 a second and hold a
>>> PersistentAppendingBuffer with those objects appended for each message?
>>> Then
>>> when the 1001:st message comes or a timetrigger kicks in, I would write
>>> everything to disk and delete the state. Would the appendingBuffer be
>>> de-/serialized for each function invocation?
>>>
>>> If yes, is there any workaround for this so the data is just held in RAM?
>>>
>>> Thanks,
>>>
>>> Regards
>>> Dan
>>>
>>>
>>>
>>> --
>>> Sent from:
>>> http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/
>>>
>>

Re: Initialization of a Stateful function

[Dan Pettersson]

Hi Igal,

If you have the time, the easiest way to troubleshoot and recreate this
race condition is to run class FlowTest in repo:
https://github.com/danp11/stateful-functions  (I've upgraded to Java11)

I've created a "stockmarket-example" under examples/ directory and the test
is running via Harness and it expected
to print "Amazon" and "Facebook" from the class FnCacheInstrument if it
runs correctly.

Running this test a few times should result in the error "There is already
a catch all case for class"   but when changing
StatefulMatchFunction to include the double checked locking it works
every time. I have a class hierarchy that might cause
this problem but it would be very interesting to know if I should model the
function in different (correct) way.

Let me know if you want me to write down more details what I'm trying to
build and if there are better ways to structure the code.

Thanks for taking the time to help out!

Regards
Dan

Den sön 12 jan. 2020 kl 09:11 skrev Igal Shilman :

> Hi Dan,
>
> The initialization and usage of a function is handled by the runtime by a
> single thread.
> Can you share more details? For example:
> 1. Any Stacktrace
> 2. How does your StatefulFunctionProvider looks like?
> Do you cache a single function instance, or return a new one in the
> provider ?
> 3. Are your functions keep any non PersistedValue state?
>
> Thanks,
> Igal
>
> On Saturday, January 11, 2020, Dan Pettersson 
> wrote:
>
>> Hi all,
>>
>> I have had some race problems in
>> method StatefulMatchFunction.ensureInitialized.
>>
>> The quick solution would be to introduce a double checked locking:
>>
>> private void ensureInitialized() {
>> if (!setup) {
>>   synchronized (this) {
>> if (!setup) {
>>   configure(matcher);
>>   setup = true;
>> }
>>   }
>> }
>>   }
>>
>> But this solution would force us to make the setup variable to volatile,
>> which would/can decrease the read performance in
>> this performance critical code..
>>
>> Any thoughts on how to solve this in a clever way would be appreciated.
>>
>> Regards
>> Dan
>>
>

Initialization of a Stateful function

[Dan Pettersson]

Hi all,

I have had some race problems in
method StatefulMatchFunction.ensureInitialized.

The quick solution would be to introduce a double checked locking:

private void ensureInitialized() {
if (!setup) {
  synchronized (this) {
if (!setup) {
  configure(matcher);
  setup = true;
}
  }
}
  }

But this solution would force us to make the setup variable to volatile,
which would/can decrease the read performance in this performance critical
code..

Any thoughts on how to solve this in a clever way would be appreciated.

Regards
Dan

Re: Abstract classes in Stateful functions

[Dan Pettersson]

Thanks!

Den tis 7 jan. 2020 12:49Igal Shilman  skrev:

> Hi Dan,
> Yes. You should be able to store “wheels” in “Vehicle”.
>
> Igal.
>
> On Monday, January 6, 2020, Dan Pettersson 
> wrote:
>
>> Hello again :-)
>>
>> When using an abstract class should the instance variables be stored in
>> PersistenceValues
>> to conform to the fault tolerance?
>>
>> Exempel Car, Truck and Abstract Vehicle, should wheels in Vehicle be
>> stored in a PersistenceValue?
>>
>> Thanks,
>>
>> /Dan
>>
>

Abstract classes in Stateful functions

[Dan Pettersson]

Hello again :-)

When using an abstract class should the instance variables be stored in
PersistenceValues
to conform to the fault tolerance?

Exempel Car, Truck and Abstract Vehicle, should wheels in Vehicle be stored
in a PersistenceValue?

Thanks,

/Dan

Re: Stateful functions and modules

[Dan Pettersson]

Ok, good. Thanks for your response.

/Dan

Den sön 5 jan. 2020 11:52Igal Shilman  skrev:

> Hi Dan,
>
> Having a class that defines only the function types indeed makes sense,
> this would lower the coupling between the (maven) module that contains the
> function implementation and the (maven) module that uses it.
> You can peek here for example:
>
>
> https://github.com/ververica/stateful-functions/blob/master/stateful-functions-examples/stateful-functions-shopping-cart-example/src/main/java/com/ververica/statefun/examples/shoppingcart/Identifiers.java#L28
>
> Igal.
>
> On Saturday, January 4, 2020, Dan Pettersson 
> wrote:
>
>> Hi,
>>
>> What is the preferred way to expose functions (FunctionTypes) between
>> modules?
>>
>> For example lets say i have the following maven modules:
>>
>> -> common
>> -> statistics
>> -> persister
>> -> tradematching
>>
>> and I have some FunctionTypes in common and persister module that should
>> be reach
>> from all modules.
>>
>> common module is included everywhere so I thought about creating
>> a helper class, CallableFunctions, and there specify the FunctionTypes
>> that can
>> be called from all modules. Is this the right approach or is there a
>> better way?
>>
>> Regards
>> Dan
>>
>

Stateful functions and modules

[Dan Pettersson]

Hi,

What is the preferred way to expose functions (FunctionTypes) between
modules?

For example lets say i have the following maven modules:

-> common
-> statistics
-> persister
-> tradematching

and I have some FunctionTypes in common and persister module that should be
reach
from all modules.

common module is included everywhere so I thought about creating
a helper class, CallableFunctions, and there specify the FunctionTypes that
can
be called from all modules. Is this the right approach or is there a better
way?

Regards
Dan

Re: Stateful function metrics

[Dan Pettersson]

Hi Igal and Thanks for your quick response and yes, you got my second
question right.

I'm a building a small PoC around fraudulent trades and in short, I've
fine-grained the
functions to the level participantId + "::" + instrumentId (ie
"BankA::AMAZON")

In this flow of stock exchange messages, there are messages that tells the
market
if the instrument has opened, is being halted or being closed for the day.

These messages come on instrumentId level so I have to route these message
to all functions with the key participantId + "::" + (actual)instrumentId.
I had hoped to be able to get a copy of all functions from the repository
to loop thru them and dispatch but I can't find a way to get hold of them.
Is there any way I can get them?

I haven't studied the core functionality enough but could it be an option
to open up the repository and return a copy of the
ObjectOpenHashMap that holds all the functions? I guess it's not a common
requirement so to keep them hidden is probably the best option.

As a workaround, I've created "Function listeners" where functions can
subscribe to a certain type of message.

For example, FnIsClosingNotifier (key is instrumentId) is holding a
PersistenceValue with all the function addresses
that subscribe to an instrument closing message. The subscription is done
from other functions in the configuration by just sending
a "Protobuf empty message" and when the closing message arrives the
dispatch to the listeners is done in FnIsClosingNotifier.

Is there a better way that you can think of to implement this kind of
requirement, where one message should be sent to (on beforehand not known)
several subscribing functions.

Below is some code that hopefully describes my current implementation to
subscribe to a certain type of message.

The function that wants to be notified when the closing message arrives.
This function has the id participantId::InstrumentId

[image: image.png]

And the notifier that holds all subscribers Addresses in the
persistenceValue "listeners"
[image: image.png]
Regards
Dan

Den mån 30 dec. 2019 kl 00:50 skrev Igal Shilman :

> Hi Dan,
>
> You can learn more about Flink’s metrics system at [1]
> You would be able to either setup a reporter that would export the metrics
> to an external system, or query the metrics via the REST API, or simply use
> Flink’s web ui to obtain them.
>
> If I understand the second part of your question correctly - you have a
> persisted value in a base class, but few different function types that
> derive from that base class, and you are wondering what is the scope of
> that persisted value?
> If that is the question, then the scope is bound to the function
> address(type+id) and not to the Java instance.
> So it is safe.
>
> [1] -
> https://ci.apache.org/projects/flink/flink-docs-release-1.9/monitoring/metrics.html
>
> Happy hacking,
> Igal
>
>
> On Sunday, December 29, 2019, Dan Pettersson 
> wrote:
>
>> Hi all
>>
>> I'm trying to get hold of some metrics from the functions that I have
>> created but can't find a way to get them. It's the metrics mentioned here
>> I'm interested about:
>> https://statefun.io/deployment_operations/metrics.html
>> Any suggestions are appreciated.
>>
>> I also have a question regarding "best practice" when dealing with
>> function that share business logic. Is it safe storage wise to extends an
>> Abstract class that holds the persistent values?
>>
>> Thanks in advance and Happy coding during the holidays :-)
>>
>> Regards
>> Dan
>>
>

Stateful function metrics

[Dan Pettersson]

Hi all

I'm trying to get hold of some metrics from the functions that I have
created but can't find a way to get them. It's the metrics mentioned here
I'm interested about:
https://statefun.io/deployment_operations/metrics.html
Any suggestions are appreciated.

I also have a question regarding "best practice" when dealing with function
that share business logic. Is it safe storage wise to extends an Abstract
class that holds the persistent values?

Thanks in advance and Happy coding during the holidays :-)

Regards
Dan

Stateful functions

[Dan Pettersson]

Hello, 

I've started to play around with Stateful functions and I like it a lot :-)
Also Thanks for the comprehensive documentation and your very good talk Igal.

I would appreciate if you could give some hints/ideas over how to structure
an application with the following criteria:

One kafka ingress with billions of incoming messages per day.
These are messages from stock exchanges and id is the stock id.

There are around 30 independent functions that will subscribe to these messages.

Is it better to only have one module that via its router sends each message 
(async)
to all the other functions or is it better that each module subscribes to the 
same
kafka ingress? With the first solution, only one deserialization will be done 
per message
but there is only one checkpointing for all the 30 functions.. With the second 
solution, 
there will be a more fine-grained fault tolerance and I guess one can 
deliver/patch modules
independently but the deserialization for each module will be an overhead 
compared to
solution 1.

I've just started learning about Flink and its Table and SQL API for the last 6 
months and now the last month about Stateful functions. So sorry if my 
questions are unclear but I would really appreciate if someone could give some 
short advice on how to structure an application as described above. Throughput 
is important and not so much the ability to restart with check-/savepoints. If 
having only one router for all functions is the best option how can one 
register each Function Type to the global router in an elegant way? 

Any guidance would be helpful so Thanks in advance,

Regards
Dan