Question about Writing Incremental Graph Algorithms using Apache Flink Gelly

[Kaan Sancak]

Hi all,
As I stated in the post here 
(https://stackoverflow.com/questions/61164302/writing-incremental-graph-algorithms-using-apache-flink-gelly
 
),
 I am trying to write a dynamic algorithm using Flink Gelly. However, I 
couldn’t find any examples. Most of the examples are performed on static 
graphs. What I need is an incremental use-case/example. I would be grateful, if 
you can point out some examples, or any help is appreciated. 

Best
Kaan Sancak

PhD Student
Georgia Tech

Re: TypeInformation composition ?

[Laurent Exsteens]

Thanks a lot!

On Sat, 11 Apr 2020 at 20:45, Seth Wiesman  wrote:

> If the type information for T is stored in a member variable called
> myTypeInfo you can do something like this.
>
> import org.apache.flink.api.common.typeinfo.Types;
>
> Types.TUPLE(Types.LONG, myTypeInfo);
>
> Seth
>
> On Apr 11, 2020, at 11:06 AM, Laurent Exsteens <
> laurent.exste...@euranova.eu> wrote:
>
> 
> Hello,
>
> I have a generic ProcessFunction using list state, for which I receive the
> type information as constructor parameter (since it is not possible to
> create the type information in the class due to type Erasure).
>
> I now need to keep not only the data, but also the timestamp at which they
> appear. But I don't find a way to create a list state of Tuple2
> (where T is the generic type, for which I receive the type information as
> parameter.
>
> I would like to create that state descriptor getting only those
> informations (in order to keep the interface of the class simple.
>
> Right now I manage by creating 2 state: one for T and one for Long.
>
> But I wonder if I could create that composite state,. For exemple n'y
> composing the type informations, but I didn't find a way to do that.
>
> Any other idea is also welcome.
>
> Thanks in advance.
>
> Regards,
>
> Laurent.
>
> ♻ Be green, keep it on the screen
>
>

-- 
*Laurent Exsteens*
Data Engineer
(M) +32 (0) 486 20 48 36

*EURA NOVA*

Rue Emile Francqui, 4

1435 Mont-Saint-Guibert

(T) +32 10 75 02 00

*euranova.eu *

*research.euranova.eu* 

-- 
♻ Be green, keep it on the screen

Re: TypeInformation composition ?

[Seth Wiesman]

If the type information for T is stored in a member variable called myTypeInfo 
you can do something like this.

import org.apache.flink.api.common.typeinfo.Types;

Types.TUPLE(Types.LONG, myTypeInfo);

Seth 

> On Apr 11, 2020, at 11:06 AM, Laurent Exsteens  
> wrote:
> 
> 
> Hello,
> 
> I have a generic ProcessFunction using list state, for which I receive the 
> type information as constructor parameter (since it is not possible to create 
> the type information in the class due to type Erasure).
> 
> I now need to keep not only the data, but also the timestamp at which they 
> appear. But I don't find a way to create a list state of Tuple2 
> (where T is the generic type, for which I receive the type information as 
> parameter.
> 
> I would like to create that state descriptor getting only those informations 
> (in order to keep the interface of the class simple.
> 
> Right now I manage by creating 2 state: one for T and one for Long.
> 
> But I wonder if I could create that composite state,. For exemple n'y 
> composing the type informations, but I didn't find a way to do that.
> 
> Any other idea is also welcome.
> 
> Thanks in advance.
> 
> Regards,
> 
> Laurent.
> 
> ♻ Be green, keep it on the screen

TypeInformation composition ?

[Laurent Exsteens]

Hello,

I have a generic ProcessFunction using list state, for which I receive the
type information as constructor parameter (since it is not possible to
create the type information in the class due to type Erasure).

I now need to keep not only the data, but also the timestamp at which they
appear. But I don't find a way to create a list state of Tuple2
(where T is the generic type, for which I receive the type information as
parameter.

I would like to create that state descriptor getting only those
informations (in order to keep the interface of the class simple.

Right now I manage by creating 2 state: one for T and one for Long.

But I wonder if I could create that composite state,. For exemple n'y
composing the type informations, but I didn't find a way to do that.

Any other idea is also welcome.

Thanks in advance.

Regards,

Laurent.

-- 
♻ Be green, keep it on the screen

Re: Setting different idleStateRetentionTime for different queries executed in the same TableEnvironment in Flink 1.10

[Jark Wu]

Yes, that's right. Set idleStateRetentionTime on TableConfig before
translation should work.

On Sat, 11 Apr 2020 at 14:46, Jiahui Jiang  wrote:

> Thank you for answering! I was reading
> StreamExecutionEnvironmentImpl/StreamPlanner, and it seems to me that when
> trying to convert tables to DataStreams, planner.translate is taking the
> current tableConfig into account (aa in it reads the current tableConfig
> content even though it’s not explicitly passed in as an argument for
> translate). So seems like if I set tableConfig right before converting to
> DataStreams that should work?
>
> Or did you mean the actual tableEnvironment.execute()? Since we have a
> whole pipeline with multiple queries that also depends on each other. We
> have to have all the continuous queries executing concurrently.
>
> Thanks again!
> --
> *From:* Jark Wu 
> *Sent:* Saturday, April 11, 2020 1:24 AM
> *To:* Jiahui Jiang 
> *Cc:* user@flink.apache.org 
> *Subject:* Re: Setting different idleStateRetentionTime for different
> queries executed in the same TableEnvironment in Flink 1.10
>
> Hi Jiahui,
>
> QueryConfig is deprecated and will be removed in the future, because it is
> confusing that TableAPI has so many different config classes.
> If you want to set different idleStateRetentionTime for different queries,
> you can set a new idleStateRetentionTime on TableConfig before
> execute/submit the query.
>
> Best,
> Jark
>
> On Sat, 11 Apr 2020 at 09:21, Jiahui Jiang 
> wrote:
>
> Just looked into the source code a bit further and realized that for
> StreamTableEnvironmentImpl, even for sinks it's also doing translation
> lazily. Any way we can have different transformation to have different
> queryConfig?
> --
> *From:* Jiahui Jiang 
> *Sent:* Friday, April 10, 2020 6:46 PM
> *To:* user@flink.apache.org 
> *Subject:* Setting different idleStateRetentionTime for different queries
> executed in the same TableEnvironment in Flink 1.10
>
> Hello! I'm using Table API to write a pipeline with multiple queries. And
> I want to set up different idleStateRetentionTime for different queries.
>
> In Flink 1.8, it seems to be the case where I can pass in a
> streamQueryConfig when converting each output table into datastreams. And
> the translate with take the idleStateRetentionTime into account.
>
> But in Flink 1.10, that idleStateRetentionTime actually gets set on
> TableConfig and applies to the tableEnvironment.
>
> Is there a way to have different idleStateRetentionTime for different
> queries in 1.10?
>
> I saw tableEnvironment.insertInto(sink, queryConfig) still allows eager
> translate. But does that mean if I have multiple sinks for the same
> datastream with different idleStateRetentionTime(s) configuration, that
> will cause the transformation to be executed multiple times?
>
> Thank you!
>
>