flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Chakravarthy varaga <chakravarth...@gmail.com>
Subject Re: Sharing Java Collections within Flink Cluster
Date Thu, 22 Sep 2016 19:22:25 GMT
Hi Team,

      Will you be able to guide me on this? Is this a known issue with
checkpointing ?

CVP

On 22 Sep 2016 15:57, "Chakravarthy varaga" <chakravarthyvp@gmail.com>
wrote:

> PFA, Flink_checkpoint_time.png in relation to this issue.
>
> On Thu, Sep 22, 2016 at 3:38 PM, Chakravarthy varaga <
> chakravarthyvp@gmail.com> wrote:
>
>> Hi Aljoscha & Fabian,
>>
>>     Finally I got this working. Thanks for your help. In terms persisting
>> the state (for S2), I tried to use checkpoint every 10 Secs using a
>> FsStateBackend... What I notice is that the checkpoint duration is  almost
>> 2 minutes for many cases, while for the other cases it varies from 100 ms
>> to 1.5 minutes frequently.
>>
>>     The pseudocode is as below:
>>
>>      KeyedStream<String, String> ks1 = ds1.keyBy("*") ;
>>      KeyedStream<Tuple2<String, V>, String> ks2 = ds2.flatMap(split T
>> into k-v pairs).keyBy(0);
>>
>>      ks1.connect(ks2).flatMap(X);
>>      //X is a CoFlatMapFunction that inserts and removes elements from
>> ks2 into a key-value state member. Elements from ks1 are matched against
>> that state.
>>
>>      //ks1 is streaming about 100K events/sec from kafka topic
>>      //ks2 is streaming about 1 event every 10 minutes... Precisely when
>> the 1st event is consumed from this stream, checkpoint takes 2 minutes
>> straightaway.
>>
>>     The version of flink is 1.1.2
>>
>>  Best Regards
>> CVP
>>
>> On Tue, Sep 13, 2016 at 7:29 AM, Aljoscha Krettek <aljoscha@apache.org>
>> wrote:
>>
>>> Hi,
>>> you don't need the BlockedEventState class, you should be able to just
>>> do this:
>>>
>>> private transient ValueState<BlockedRoadInfo> blockedRoads;
>>>          ............
>>>       @Override
>>>     public void open(final org.apache.flink.configuration.Configuration
>>> parameters) throws Exception {
>>>         final ValueStateDescriptor<BlockedRoadInfo> blockedStateDesc =
>>> new ValueStateDescriptor<BlockedRoadInfo>("BlockedEventStates",
>>>                 TypeInformation.of(BlockedRoadInfo.class), null);
>>>         blockedRoads = getRuntimeContext().getState(blockedStateDesc);
>>>
>>>     };
>>>
>>>   }
>>>
>>> Cheers,
>>> Aljoscha
>>>
>>>
>>> On Mon, 12 Sep 2016 at 16:24 Chakravarthy varaga <
>>> chakravarthyvp@gmail.com> wrote:
>>>
>>>> Hi Fabian,
>>>>
>>>>     I'm coding to check if your proposal works and hit with an issue
>>>> with ClassCastException.
>>>>
>>>>
>>>>     // Here is my Value that has state information.....an
>>>> implementation of my value state... where the key is a Double value... on
>>>> connected stream ks2
>>>>
>>>>     public class BlockedEventState implements
>>>> ValueState<BlockedRoadInfo> {
>>>>
>>>>     public BlockedRoadInfo blockedRoad;
>>>>
>>>>     @Override
>>>>     public void clear() {
>>>>         blockedRoad = null;
>>>>
>>>>     }
>>>>
>>>>     @Override
>>>>     public BlockedRoadInfo value() throws IOException {
>>>>         return blockedRoad;
>>>>     }
>>>>
>>>>     @Override
>>>>     public void update(final BlockedRoadInfo value) throws IOException {
>>>>         blockedRoad = value;
>>>>     }
>>>> }
>>>>
>>>>        //BlockedRoadInfo class...
>>>>         public class BlockedRoadInfo {
>>>>             long maxLink;
>>>>             long minLink;
>>>>             double blockedEventId;
>>>>     ....setters & ... getters
>>>> }
>>>>
>>>> /// new RichCoFlatMapFunction() {
>>>>
>>>> private transient BlockedEventState blockedRoads;
>>>>          ............
>>>>       @Override
>>>>     public void open(final org.apache.flink.configuration.Configuration
>>>> parameters) throws Exception {
>>>>         final ValueStateDescriptor<BlockedRoadInfo> blockedStateDesc
=
>>>> new ValueStateDescriptor<BlockedRoadInfo>("BlockedEventStates",
>>>>                 TypeInformation.of(BlockedRoadInfo.class), null);
>>>>         blockedRoads = (BlockedEventState)
>>>> getRuntimeContext().getState(blockedStateDesc); * // FAILS HERE WITH
>>>> CLASSCAST*
>>>>
>>>>     };
>>>>
>>>>   }
>>>>
>>>>
>>>>
>>>>
>>>> *Caused by: java.lang.ClassCastException:
>>>> org.apache.flink.runtime.state.filesystem.FsValueState cannot be cast to
>>>> com.ericsson.components.aia.io
>>>> <http://com.ericsson.components.aia.io>t.volvo.state.BlockedEventState*
>>>>
>>>>
>>>>
>>>> *I have tried to set the state backend to both MemState and
>>>> FsState...streamEnv.setStateBackend(new
>>>> FsStateBackend("file:///tmp/flink/checkpoints"));*
>>>>
>>>>
>>>>
>>>> On Thu, Sep 8, 2016 at 10:10 PM, Fabian Hueske <fhueske@gmail.com>
>>>> wrote:
>>>>
>>>>> Not sure if I got your requirements right, but would this work?
>>>>>
>>>>> KeyedStream<String, String> ks1 = ds1.keyBy("*") ;
>>>>> KeyedStream<Tuple2<String, V>, String> ks2 = ds2.flatMap(split
T into
>>>>> k-v pairs).keyBy(0);
>>>>>
>>>>> ks1.connect(ks2).flatMap(X)
>>>>>
>>>>> X is a CoFlatMapFunction that inserts and removes elements from ks2
>>>>> into a key-value state member. Elements from ks1 are matched against
that
>>>>> state.
>>>>>
>>>>> Best, Fabian
>>>>>
>>>>> 2016-09-08 20:24 GMT+02:00 Chakravarthy varaga <
>>>>> chakravarthyvp@gmail.com>:
>>>>>
>>>>>> Hi Fabian,
>>>>>>
>>>>>>      First of all thanks for all your prompt responses. With regards
>>>>>> to 2) Multiple looks ups, I have to clarify what I mean by that...
>>>>>>
>>>>>>      DS1<String> elementKeyStream  = stream1.map(String<>);
this maps
>>>>>> each of the streaming elements into string mapped value...
>>>>>>      DS2<T>  = stream2.xxx(); // where stream2 is a kafka source
>>>>>> stream, as you proposed.. xxx() should be my function() which splits
the
>>>>>> string and generates key1:<value1>, key2:<value2>, key3:<value3>
>>>>>> ....keyN:<value4>
>>>>>>
>>>>>>      Now,
>>>>>>         I wish to map elementKeyStream with look ups within (key1,
>>>>>> key2...keyN) where key1, key2.. keyN and their respective values
should be
>>>>>> available across the cluster...
>>>>>>
>>>>>> Thanks a million !
>>>>>> CVP
>>>>>>
>>>>>>
>>>>>>
>>>>>>
>>>>>> On Wed, Sep 7, 2016 at 9:15 PM, Fabian Hueske <fhueske@gmail.com>
>>>>>> wrote:
>>>>>>
>>>>>>> That depends.
>>>>>>> 1) Growing/Shrinking: This should work. New entries can always
be
>>>>>>> inserted. In order to remove entries from the k-v-state you have
to set the
>>>>>>> value to null. Note that you need an explicit delete-value record
to
>>>>>>> trigger the eviction.
>>>>>>> 2) Multiple lookups: This does only work if all lookups are
>>>>>>> independent from each other. You can partition DS1 only on a
single key and
>>>>>>> the other keys might be located on different shards. A workaround
might be
>>>>>>> to duplicate S1 events for each key that they need to look up.
However, you
>>>>>>> might need to collect events from the same S1 event after the
join. If that
>>>>>>> does not work for you, the only thing that comes to my mind is
to broadcast
>>>>>>> the state and keep a full local copy in each operator.
>>>>>>>
>>>>>>> Let me add one more thing regarding the upcoming rescaling feature.
>>>>>>> If this is interesting for you, rescaling will also work (maybe
not in the
>>>>>>> first version) for broadcasted state, i.e. state which is the
same on all
>>>>>>> parallel operator instances.
>>>>>>>
>>>>>>> 2016-09-07 21:45 GMT+02:00 Chakravarthy varaga <
>>>>>>> chakravarthyvp@gmail.com>:
>>>>>>>
>>>>>>>> I'm understanding this better with your explanation..
>>>>>>>> With this use case,    each element in DS1 has to look up
against a
>>>>>>>> 'bunch of keys' from DS2 and DS2 could shrink/expand in terms
of the no.,
>>>>>>>> of keys.... will the key-value shard work in this case?
>>>>>>>>
>>>>>>>> On Wed, Sep 7, 2016 at 7:44 PM, Fabian Hueske <fhueske@gmail.com>
>>>>>>>> wrote:
>>>>>>>>
>>>>>>>>> Operator state is always local in Flink. However, with
key-value
>>>>>>>>> state, you can have something which behaves kind of similar
to a distribute
>>>>>>>>> hashmap, because each operator holds a different shard/partition
of the
>>>>>>>>> hashtable.
>>>>>>>>>
>>>>>>>>> If you have to do only a single key lookup for each element
of
>>>>>>>>> DS1, you should think about partitioning both streams
(keyBy) and writing
>>>>>>>>> the state into Flink's key-value state [1].
>>>>>>>>>
>>>>>>>>> This will have several benefits:
>>>>>>>>> 1) State does not need to be replicated
>>>>>>>>> 2) Depending on the backend (RocksDB) [2], parts of the
state can
>>>>>>>>> reside on disk. You are not bound to the memory of the
JVM.
>>>>>>>>> 3) Flink takes care of the look-up. No need to have your
own
>>>>>>>>> hashmap.
>>>>>>>>> 4) It will only be possible to rescale jobs with key-value
state
>>>>>>>>> (this feature is currently under development).
>>>>>>>>>
>>>>>>>>> If using the key-value state is possible, I'd go for
that.
>>>>>>>>>
>>>>>>>>> Best, Fabian
>>>>>>>>>
>>>>>>>>> [1] https://ci.apache.org/projects/flink/flink-docs-release-1.1/
>>>>>>>>> apis/streaming/state.html
>>>>>>>>> [2] https://ci.apache.org/projects/flink/flink-docs-release-1.1/
>>>>>>>>> apis/streaming/state_backends.html
>>>>>>>>>
>>>>>>>>> 2016-09-07 19:55 GMT+02:00 Chakravarthy varaga <
>>>>>>>>> chakravarthyvp@gmail.com>:
>>>>>>>>>
>>>>>>>>>> certainly, what I thought as well...
>>>>>>>>>> The output of DataStream2 could be in 1000s and there
are state
>>>>>>>>>> updates...
>>>>>>>>>> reading this topic from the other job, job1, is okie.
>>>>>>>>>> However, assuming that we maintain this state into
a collection,
>>>>>>>>>> and updating the state (by reading from the topic)
in this collection, will
>>>>>>>>>> this be replicated across the cluster within this
job1 ?
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>> On Wed, Sep 7, 2016 at 6:33 PM, Fabian Hueske <fhueske@gmail.com>
>>>>>>>>>> wrote:
>>>>>>>>>>
>>>>>>>>>>> Is writing DataStream2 to a Kafka topic and reading
it from the
>>>>>>>>>>> other job an option?
>>>>>>>>>>>
>>>>>>>>>>> 2016-09-07 19:07 GMT+02:00 Chakravarthy varaga
<
>>>>>>>>>>> chakravarthyvp@gmail.com>:
>>>>>>>>>>>
>>>>>>>>>>>> Hi Fabian,
>>>>>>>>>>>>
>>>>>>>>>>>>     Thanks for your response. Apparently
these DataStream
>>>>>>>>>>>> (Job1-DataStream1 & Job2-DataStream2)
are from different flink applications
>>>>>>>>>>>> running within the same cluster.
>>>>>>>>>>>>     DataStream2 (from Job2) applies transformations
and updates
>>>>>>>>>>>> a 'cache' on which (Job1) needs to work on.
>>>>>>>>>>>>     Our intention is to not use the external
key/value store as
>>>>>>>>>>>> we are trying to localize the cache within
the cluster.
>>>>>>>>>>>>     Is there a way?
>>>>>>>>>>>>
>>>>>>>>>>>> Best Regards
>>>>>>>>>>>> CVP
>>>>>>>>>>>>
>>>>>>>>>>>> On Wed, Sep 7, 2016 at 5:00 PM, Fabian Hueske
<
>>>>>>>>>>>> fhueske@gmail.com> wrote:
>>>>>>>>>>>>
>>>>>>>>>>>>> Hi,
>>>>>>>>>>>>>
>>>>>>>>>>>>> Flink does not provide shared state.
>>>>>>>>>>>>> However, you can broadcast a stream to
CoFlatMapFunction, such
>>>>>>>>>>>>> that each operator has its own local
copy of the state.
>>>>>>>>>>>>>
>>>>>>>>>>>>> If that does not work for you because
the state is too large
>>>>>>>>>>>>> and if it is possible to partition the
state (and both streams), you can
>>>>>>>>>>>>> also use keyBy instead of broadcast.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Finally, you can use an external system
like a KeyValue Store
>>>>>>>>>>>>> or In-Memory store like Apache Ignite
to hold your distributed collection.
>>>>>>>>>>>>>
>>>>>>>>>>>>> Best, Fabian
>>>>>>>>>>>>>
>>>>>>>>>>>>> 2016-09-07 17:49 GMT+02:00 Chakravarthy
varaga <
>>>>>>>>>>>>> chakravarthyvp@gmail.com>:
>>>>>>>>>>>>>
>>>>>>>>>>>>>> Hi Team,
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>      Can someone help me here? Appreciate
any response !
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> Best Regards
>>>>>>>>>>>>>> Varaga
>>>>>>>>>>>>>>
>>>>>>>>>>>>>> On Mon, Sep 5, 2016 at 4:51 PM, Chakravarthy
varaga <
>>>>>>>>>>>>>> chakravarthyvp@gmail.com> wrote:
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Hi Team,
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>     I'm working on a Flink Streaming
application. The data
>>>>>>>>>>>>>>> is injected through Kafka connectors.
The payload volume is roughly
>>>>>>>>>>>>>>> 100K/sec. The event payload is
a string. Let's call this as DataStream1.
>>>>>>>>>>>>>>> This application also uses another
DataStream, call it
>>>>>>>>>>>>>>> DataStream2, (consumes events
off a kafka topic). The elements of this
>>>>>>>>>>>>>>> DataStream2 involves in a certain
transformation that finally updates a
>>>>>>>>>>>>>>> Hashmap(/Java util Collection).
Apparently the flink application should
>>>>>>>>>>>>>>> share this HashMap across the
flink cluster so that DataStream1 application
>>>>>>>>>>>>>>> could check the state of the
values in this collection. Is there a way to
>>>>>>>>>>>>>>> do this in Flink?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>>     I don't see any Shared Collection
used within the
>>>>>>>>>>>>>>> cluster?
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>> Best Regards
>>>>>>>>>>>>>>> CVP
>>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>>
>>>>>>>>>>>>>
>>>>>>>>>>>>
>>>>>>>>>>>
>>>>>>>>>>
>>>>>>>>>
>>>>>>>>
>>>>>>>
>>>>>>
>>>>>
>>>>
>>
>

Mime
View raw message