[jira] [Commented] (SPARK-2629) Improved state management for Spark Streaming

["Iain Cundy (JIRA)" <ji...@apache.org>]

    [ https://issues.apache.org/jira/browse/SPARK-2629?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15184990#comment-15184990 ] 

Iain Cundy commented on SPARK-2629:
-----------------------------------

Hi Rod

I have a similar requirement - the supported notion of a timeout from last seen is useful for some cases, but is too simple for our needs. I thought from reading blogs that maybe if the timeout was set it would happen, but it seems not - my empirical conclusion is that with timeout set the call function is invoked for keys with new state and keys that are timing out, but not any others that have no new state.

I also have a problem and possible bug - the timed out state is being removed from the list of current keys, but it is never being removed by compaction. – the size of the state and the time taken to iterate over it for each micro-batch keeps increasing over time, long after the number of ‘current’ keys settles down. We start removing keys after just over an hour, but the size of the state keeps increasing in runs of over 6 hours.

Essentially we start by just adding keys for our input tuples, reaching a peak of about 7 million keys. Then we start to output data and remove keys – the number of keys drops to about 5 million. We continue processing tuples, which adds keys, while removing the keys we no longer need – the number of keys fluctuates up and down between 5 million and  8 million. 

We know this, and are reasonably confident the timing out removal of keys is correct, because we obtain the state with JavaMapWithStateDStream.stateSnapshots and count the keys.

>From my reading (I don’t know scala!) of the code in org.apache.spark.streaming.util.StateMap.scala it seems clear that the removed keys are only marked as deleted and are really destroyed subsequently by compaction, based upon the length of the chain of delta maps. We’d expect the size of the state RDDs and the time taken to iterate over all the state to stabilize once compaction is run after we remove keys, but it just doesn’t happen.

Is there some possible reason why compaction never gets run?

I tried to use the (undocumented?) config setting spark.streaming.sessionByKey.deltaChainThreshold to try to control how often compaction is run with:
--conf spark.streaming.sessionByKey.deltaChainThreshold=2

I can see it in the Spark application UI Environment page, but it doesn’t seem to make any difference.  

I have noticed that the timeout mechanism only gets invoked on every 10th micro-batch. I’m almost sure it isn’t a coincidence that the checkpoint interval is also 10 micro-batches. I assume that is an intentional performance optimization. However because I have a lot of keys, I have a large micro-batch duration, so it would make sense for me to reduce that factor of 10. However, since I don’t call checkpoint on the state stream I can’t see how to change it?  

Can I change the checkpoint interval  somewhere? [I tried calling JavaMapWithStateDStream.checkpoint myself, but that evidently isn’t the same thing!] 

My initial assumption was that there is a new deltaMap for each micro-batch, but having noticed the timeout behavior I wonder if there is only a new deltaMap for each checkpoint? Or maybe there are other criteria?

Perhaps compaction just hasn’t run before my application falls over? Can anyone clarify exactly when it should run?

Or maybe compaction doesn’t delete old removed keys for some reason?

Any clarification of the implementation semantics would be much appreciated.

Cheers
Iain

> Improved state management for Spark Streaming
> ---------------------------------------------
>
>                 Key: SPARK-2629
>                 URL: https://issues.apache.org/jira/browse/SPARK-2629
>             Project: Spark
>          Issue Type: Epic
>          Components: Streaming
>    Affects Versions: 0.9.2, 1.0.2, 1.2.2, 1.3.1, 1.4.1, 1.5.1
>            Reporter: Tathagata Das
>            Assignee: Tathagata Das
>
>  Current updateStateByKey provides stateful processing in Spark Streaming. It allows the user to maintain per-key state and manage that state using an updateFunction. The updateFunction is called for each key, and it uses new data and existing state of the key, to generate an updated state. However, based on community feedback, we have learnt the following lessons.
> - Need for more optimized state management that does not scan every key
> - Need to make it easier to implement common use cases - (a) timeout of idle data, (b) returning items other than state
> The high level idea that I am proposing is 
> - Introduce a new API -trackStateByKey- *mapWithState* that, allows the user to update per-key state, and emit arbitrary records. The new API is necessary as this will have significantly different semantics than the existing updateStateByKey API. This API will have direct support for timeouts.
> - Internally, the system will keep the state data as a map/list within the partitions of the state RDDs. The new data RDDs will be partitioned appropriately, and for all the key-value data, it will lookup the map/list in the state RDD partition and create a new list/map of updated state data. The new state RDD partition will be created based on the update data and if necessary, with old data. 
> Here is the detailed design doc (*outdated, to be updated*). Please take a look and provide feedback as comments.
> https://docs.google.com/document/d/1NoALLyd83zGs1hNGMm0Pc5YOVgiPpMHugGMk6COqxxE/edit#heading=h.ph3w0clkd4em



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

---------------------------------------------------------------------
To unsubscribe, e-mail: issues-unsubscribe@spark.apache.org
For additional commands, e-mail: issues-help@spark.apache.org