Incremental updates
Locked
Incremental updates
 
|
Re: Incremental updates
|
|
Hi Gosia
You can have a look to the PROTEUS project we are doing [1]. We are implementing incremental version of analytics operations. For example you can see in [2] the implementation of the incremental AVG. Maybe the code can give you some ideas :-) [1] https://github.com/proteus-h2020/proteus-backend/tree/development [2] https://github.com/proteus-h2020/proteus-backend/blob/development/src/main/java/com/treelogic/proteus/flink/incops/IncrementalAverage.java ______________________________________ Dr. Rubén Casado Head of Big Data Treelogic    ruben.casado.treelogic+34 902 286 386 - +34 607 18 28 06 Parque Tecnológico de Asturias · Parcela 30 E33428 Llanera · Asturias [Spain] www.treelogic.com ______________________________________ ----- Mensaje original ----- De: "Malgorzata Kudelska" <[hidden email]> Para: [hidden email] Enviados: Martes, 24 de Mayo 2016 22:01:28 GMT +01:00 Amsterdam / Berlín / Berna / Roma / Estocolmo / Viena Asunto: Incremental updates Hi, In particular, I have a custom StateValue object and during the checkpoints I would like to save only the fields that changed since the previous checkpoint. Is that possible? Regards, |
Re: Incremental updates
|
|
Hi Gosia, right now, Flink is not doing incremental checkpoints. Every checkpoint is fully valid in isolation. Incremental checkpointing came up several times on ML discussions and we a planning to work on it once someone finds some free time. Cheers, Aljoscha On Wed, 25 May 2016 at 09:29 Rubén Casado <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, Thanks for your reply. Is Flink able to detect that an additional server joined and rebalance the processing? How is it done if I have a keyed stream and some custom ValueState variables? Cheers, Gosia 2016-05-25 11:32 GMT+02:00 Aljoscha Krettek <[hidden email]>:
|
Re: Incremental updates
|
|
Hi,
right now, this does not work but we're is also actively working on that. This is the design doc for part one of the necessary changes: https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit?usp=sharing Cheers, Aljoscha On Wed, 25 May 2016 at 13:32 Malgorzata Kudelska <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, Cheers, Hi,
right now, this does not work but we're is also actively working on that. This is the design doc for part one of the necessary changes: https://docs.google.com/document/d/1G1OS1z3xEBOrYD4wSu-LuBCyPUWyFd9l3T9WyssQ63w/edit?usp=sharing Cheers, Aljoscha On Wed, 25 May 2016 at 13:32 Malgorzata Kudelska <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, first question: are you manually keying by "userId % numberOfPartitions"? Flink internally does roughly "key.hash() % numPartitions" so it is enough to specify the userId as your key. Now, for you questions: 1. What Flink guarantees is that the state for a key k is always available when an element with key k is being processed. Internally, this means that elements with the same key will be processed by the same partition, though there would be other ways of achieving those guarantees. 2. Right now, when a node disappears the job will fail. Then recovery will kick in and restore from the latest checkpoint on a (possibly) new set of nodes. The system will make sure that the partitions and the state are correctly matched. 3. Also answered by the above, I hope at least :-) 4. This does currently not work but the ongoing work in this is tracked by https://issues.apache.org/jira/browse/FLINK-3755. Cheers, Aljoscha On Wed, 25 May 2016 at 21:09 Malgorzata Kudelska <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, Cheers, 25 maj 2016 22:54 "Aljoscha Krettek" <[hidden email]> napisał(a):
|
Re: Incremental updates
|
|
Hi, newly added nodes would sit idle, yes. Only when we finish the rescaling work mentioned in the link will we be able to dynamically adapt. The internal implementation of this will in fact hash keys to a larger number of partitions than the number of individual partitions and use these "key groups" to allows scaling to differing numbers of partitions. Once this is in it will also work on Yarn. Right now, running on Yarn does not allow a job to dynamically pick up new computing resources. Cheers, Aljoscha On Thu, 26 May 2016 at 15:50 Malgorzata Kudelska <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, Cheers, Hi, newly added nodes would sit idle, yes. Only when we finish the rescaling work mentioned in the link will we be able to dynamically adapt. The internal implementation of this will in fact hash keys to a larger number of partitions than the number of individual partitions and use these "key groups" to allows scaling to differing numbers of partitions. Once this is in it will also work on Yarn. Right now, running on Yarn does not allow a job to dynamically pick up new computing resources. Cheers, Aljoscha On Thu, 26 May 2016 at 15:50 Malgorzata Kudelska <[hidden email]> wrote:
|
Re: Incremental updates
|
|
Hi, the state will be kept indefinitely but we are planning to introduce a setting that would allow setting a time-to-live on state. I think this is exactly what you would need. As an alternative, maybe you could implement your program using windows? In this way you would also bound how long state is kept. As for using ValueState, here's a short example: DataStream<MyType> stream = ...; KeyedStream<MyType> keyedStream = stream.keyBy("id"); keyedStream.map(new RichMapFunction<MyType, Tuple2<MyType, Long>>() { private ValueState<Long> count; public void open(Configuration cfg) { state = getRuntimeContext().getState( new ValueStateDescriptor<Long>("count", LongSerializer.INSTANCE, 0L)); } public Tuple2<MyType, Long> map(MyType value) { long count = state.value() + 1; state.update(value); return new Tuple2<>(value, count); } }); Best, Aljoscha On Fri, 27 May 2016 at 18:59 Malgorzata Kudelska <[hidden email]> wrote:
|
| Free forum by Nabble | Edit this page |