Shixiong has already opened the PR - https://github.com/apache/spark/pull/11081
On Thu, Feb 4, 2016 at 11:11 AM, Yuval Itzchakov <yuva...@gmail.com> wrote: > Let me know if you do need a pull request for this, I can make that happen > (given someone does a vast PR to make sure I'm understanding this problem > right). > > On Thu, Feb 4, 2016 at 8:21 PM, Shixiong(Ryan) Zhu < > shixi...@databricks.com> wrote: > >> Thanks for reporting it. I will take a look. >> >> On Thu, Feb 4, 2016 at 6:56 AM, Yuval.Itzchakov <yuva...@gmail.com> >> wrote: >> >>> Hi, >>> I've been playing with the expiramental PairDStreamFunctions.mapWithState >>> feature and I've seem to have stumbled across a bug, and was wondering if >>> anyone else has been seeing this behavior. >>> >>> I've opened up an issue in the Spark JIRA, I simply want to pass this >>> along >>> in case anyone else is experiencing such a failure or perhaps someone has >>> insightful information if this is actually a bug: SPARK-13195 >>> <https://issues.apache.org/jira/browse/SPARK-13195> >>> >>> Using the new spark mapWithState API, I've encountered a bug when >>> setting a >>> timeout for mapWithState but no explicit state handling. >>> >>> h1. Steps to reproduce: >>> >>> 1. Create a method which conforms to the StateSpec signature, make sure >>> to >>> not update any state inside it using *state.update*. Simply create a >>> "pass >>> through" method, may even be empty. >>> 2. Create a StateSpec object with method from step 1, which explicitly >>> sets >>> a timeout using *StateSpec.timeout* method. >>> 3. Create a DStream pipeline that uses mapWithState with the given >>> StateSpec. >>> 4. Run code using spark-submit. You'll see that the method ends up >>> throwing >>> the following exception: >>> >>> {code} >>> org.apache.spark.SparkException: Job aborted due to stage failure: Task >>> 0 in >>> stage 136.0 failed 4 times, most recent failure: Lost task 0.3 in stage >>> 136.0 (TID 176, ****): java.util.NoSuchElementException: State is not set >>> at org.apache.spark.streaming.StateImpl.get(State.scala:150) >>> at >>> >>> org.apache.spark.streaming.rdd.MapWithStateRDDRecord$$anonfun$updateRecordWithData$1.apply(MapWithStateRDD.scala:61) >>> at >>> >>> org.apache.spark.streaming.rdd.MapWithStateRDDRecord$$anonfun$updateRecordWithData$1.apply(MapWithStateRDD.scala:55) >>> at scala.collection.Iterator$class.foreach(Iterator.scala:727) >>> at >>> >>> org.apache.spark.InterruptibleIterator.foreach(InterruptibleIterator.scala:28) >>> at >>> >>> org.apache.spark.streaming.rdd.MapWithStateRDDRecord$.updateRecordWithData(MapWithStateRDD.scala:55) >>> at >>> >>> org.apache.spark.streaming.rdd.MapWithStateRDD.compute(MapWithStateRDD.scala:154) >>> at >>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306) >>> at >>> org.apache.spark.CacheManager.getOrCompute(CacheManager.scala:69) >>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:268) >>> at >>> org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:38) >>> at >>> org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:306) >>> at org.apache.spark.rdd.RDD.iterator(RDD.scala:270) >>> at >>> org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:66) >>> at org.apache.spark.scheduler.Task.run(Task.scala:89) >>> at >>> org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:213) >>> at >>> >>> java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1142) >>> at >>> >>> java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:617) >>> at java.lang.Thread.run(Thread.java:745) >>> {code} >>> >>> h1. Sample code to reproduce the issue: >>> >>> {code:Title=MainObject} >>> import org.apache.spark.streaming._ >>> import org.apache.spark.{SparkConf, SparkContext} >>> /** >>> * Created by yuvali on 04/02/2016. >>> */ >>> object Program { >>> >>> def main(args: Array[String]): Unit = { >>> >>> val sc = new SparkConf().setAppName("mapWithState bug reproduce") >>> val sparkContext = new SparkContext(sc) >>> >>> val ssc = new StreamingContext(sparkContext, Seconds(4)) >>> val stateSpec = StateSpec.function(trackStateFunc >>> _).timeout(Seconds(60)) >>> >>> // Create a stream that generates 1000 lines per second >>> val stream = ssc.receiverStream(new DummySource(10)) >>> >>> // Split the lines into words, and create a paired (key-value) >>> dstream >>> val wordStream = stream.flatMap { >>> _.split(" ") >>> }.map(word => (word, 1)) >>> >>> // This represents the emitted stream from the trackStateFunc. Since >>> we >>> emit every input record with the updated value, >>> // this stream will contain the same # of records as the input >>> dstream. >>> val wordCountStateStream = wordStream.mapWithState(stateSpec) >>> wordCountStateStream.print() >>> >>> ssc.remember(Minutes(1)) // To make sure data is not deleted by the >>> time >>> we query it interactively >>> >>> // Don't forget to set checkpoint directory >>> ssc.checkpoint("") >>> ssc.start() >>> ssc.awaitTermination() >>> } >>> >>> def trackStateFunc(batchTime: Time, key: String, value: Option[Int], >>> state: State[Long]): Option[(String, Long)] = { >>> val sum = value.getOrElse(0).toLong + state.getOption.getOrElse(0L) >>> val output = (key, sum) >>> Some(output) >>> } >>> } >>> {code} >>> >>> {code:Title=DummySource} >>> >>> /** >>> * Created by yuvali on 04/02/2016. >>> */ >>> >>> import org.apache.spark.storage.StorageLevel >>> import scala.util.Random >>> import org.apache.spark.streaming.receiver._ >>> >>> class DummySource(ratePerSec: Int) extends >>> Receiver[String](StorageLevel.MEMORY_AND_DISK_2) { >>> >>> def onStart() { >>> // Start the thread that receives data over a connection >>> new Thread("Dummy Source") { >>> override def run() { receive() } >>> }.start() >>> } >>> >>> def onStop() { >>> // There is nothing much to do as the thread calling receive() >>> // is designed to stop by itself isStopped() returns false >>> } >>> >>> /** Create a socket connection and receive data until receiver is >>> stopped >>> */ >>> private def receive() { >>> while(!isStopped()) { >>> store("I am a dummy source " + Random.nextInt(10)) >>> Thread.sleep((1000.toDouble / ratePerSec).toInt) >>> } >>> } >>> } >>> {code} >>> >>> The given issue resides in the following >>> *MapWithStateRDDRecord.updateRecordWithData*, starting line 55, in the >>> following code block: >>> >>> {code} >>> dataIterator.foreach { case (key, value) => >>> wrappedState.wrap(newStateMap.get(key)) >>> val returned = mappingFunction(batchTime, key, Some(value), >>> wrappedState) >>> if (wrappedState.isRemoved) { >>> newStateMap.remove(key) >>> } else if (wrappedState.isUpdated || >>> timeoutThresholdTime.isDefined) >>> /* <--- problem is here */ { >>> newStateMap.put(key, wrappedState.get(), batchTime.milliseconds) >>> } >>> mappedData ++= returned >>> } >>> {code} >>> >>> In case the stream has a timeout set, but the state wasn't set at all, >>> the >>> "else-if" will still follow through because the timeout is defined but >>> "wrappedState" is empty and wasn't set. >>> >>> If it is mandatory to update state for each entry of *mapWithState*, then >>> this code should throw a better exception than "NoSuchElementException", >>> which doesn't really saw anything to the developer. >>> >>> I haven't provided a fix myself because I'm not familiar with the spark >>> implementation, but it seems to be there needs to either be an extra >>> check >>> if the state is set, or as previously stated a better exception message. >>> >>> >>> >>> >>> >>> -- >>> View this message in context: >>> http://apache-spark-user-list.1001560.n3.nabble.com/PairDStreamFunctions-mapWithState-fails-in-case-timeout-is-set-without-updating-State-S-tp26147.html >>> Sent from the Apache Spark User List mailing list archive at Nabble.com. >>> >>> --------------------------------------------------------------------- >>> To unsubscribe, e-mail: user-unsubscr...@spark.apache.org >>> For additional commands, e-mail: user-h...@spark.apache.org >>> >>> >> > > > -- > Best Regards, > Yuval Itzchakov. >
