mapWithState with a big initial RDD gets OOM'ed

[Daniel Haviv]

Hi,
I have a stateful streaming app where I pass a rather large initialState
RDD at the beginning.
No matter to how many partitions I divide the stateful stream I keep
failing on OOM or Java heap space.

Is there a way to make it more resilient?
how can I control it's storage level?

This is basically my code:

val x = ss.sql("select * From myTable where partition_ts >=
2016110600").toJSON.rdd.mapPartitions(extractIidfromJson)
val stateSpec = StateSpec.function(trackStateFunc _).numPartitions(128)
  .timeout(Durations.minutes(60 * 48)).initialState(x)
val ssc = new StreamingContext(sc, Seconds(10))
val sqlContext = ss.sqlContext


val stateStream = kafkaStream.mapPartitions(r =>
jsonToJsonNode(r)).mapWithState(stateSpec)

stateStream.foreachRDD(r=>{  if (!r.isEmpty())
{ss.read.json(r).write.format("orc").mode(SaveMode.Append).saveAsTable("joinedData")}}
)


Thank you,

Daniel