spark-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From Adrian Tanase <>
Subject Re: [Spark Streaming] How do we reset the updateStateByKey values.
Date Mon, 26 Oct 2015 10:41:08 GMT
Have you considered union-ing the 2 streams? Basically you can consider them as 2 “message
types” that your update function can consume (e.g. implement a common interface):

  *   regularUpdate
  *   resetStateUpdate

Inside your updateStateByKey you can check if any of the messages in the list of updates is
an resetState message. If now, continue summing the others.

I can provide scala samples, my java is beyond rusty :)


From: Uthayan Suthakar
Date: Friday, October 23, 2015 at 2:10 PM
To: Sander van Dijk
Cc: user
Subject: Re: [Spark Streaming] How do we reset the updateStateByKey values.

Hi Sander,

Thank you for your very informative email. From your email, I've learned a quite a bit.

>>>Is the condition determined somehow from the data coming through streamLogs, and
is newData streamLogs again (rather than a whole data source?)

No, they are two different Streams. I have two stream receivers, one of which sends event
regularly and the other is not so regular (this data is computed by another application and
stored into HDFS). What I'm trying to do is pick up the data from HDFS and overwrite the Stream's
state. Hence the overwriting should only take place if there were new files in HDFS.

So we have two different RDDs. If no file is found in HDFS, it will simply read the regular
stream, compute and update the state(1) and output the result. If there is a file found in
HDFS, then it should overwrite the state (1) with the data found from HDFS so the new events
from the regular stream will carry on with the new overwritten state.

I managed to get most of it done, but only having the issue with overwriting the state.

On 22 October 2015 at 19:35, Sander van Dijk <<>>
I don't think it is possible in the way you try to do it. It is important to remember that
the statements you mention only set up the stream stages, before the stream is actually running.
Once it's running, you cannot change, remove or add stages.

I am not sure how you determine your condition and what the actual change should be when that
condition is met: you say you want a different update function but then give a statement with
the same update function but a different source stream). Is the condition determined somehow
from the data coming through streamLogs, and is newData basically streamLogs again (rather
than a whole data source?). In that case I can think of 3 things to try:

- if the condition you switch on can be determined independently from every item in streamLogs,
you can simply do an if/else inside updateResultsStream to change the method that you determine
your state
- if this is not the case, but you can determine when to switch your condition for each key
independently, you can extend your state type to also keep track of your condition: rather
than using JavaPairDStream<String, String> you make updatedResultsState a JavaPairDStream<String,
Pair<String, Boolean>> (assuming you have some class Pair), and you make updateResultsStream
update and check the state of the boolean.
- finally, you can have a separate state stream that keeps track of your condition globally,
then join that with you main stream and use that to update state. Something like:

// determineCondition should result in a reduction to a single item that signals whether the
condition is met in the current batch, updateContitionState should remember that
conditionStateStream = streamLogs.reduce(determineCondition).updateStateByKey(updateConditionState)

// addCondition gets RDDs from streamLogs and  single-item RDDs with the condition state and
should add that state to each item in the streamLogs RDD
joinedStream = streamLogs.transformWith(conditionStateStream, addCondition)

// This is similar to the extend state type of the previous idea, but now your condition state
is determined globally rather than per log entry
updatedResultsState = joinedStream.updateStateByKey(updateResultsStream)

I hope this applies to your case and that it makes sense, my Java is a bit rusty :) and perhaps
others can suggest better spark streaming methods that can be used, but hopefully the idea
is clear.


On Thu, Oct 22, 2015 at 4:06 PM Uthayan Suthakar <<>>
Hello guys,

I have a stream job that will carryout computations and update the state (SUM the value).
At some point, I would like to reset the state. I could drop the state by setting 'None' but
I don't want to drop it. I would like to keep the state but update the state.

For example:

JavaPairDStream<String, String> updatedResultsState = streamLogs.updateStateByKey(updateResultsStream);

At some condition, I would like to update the state by key but with the different values,
hence different update function.


 updatedResultsState = newData.updateStateByKey(resetResultsStream);

But the  newData.updateStateByKeyvalues cannot be replaced with the value in streamLogs.updateStateByKey.
Do you know how I could replace the state value in  streamLogs with newData.

Is this possible?

View raw message