Can you show us the output of DStream#print() if you have it ? Thanks
On Tue, Mar 31, 2015 at 2:55 AM, Nicolas Phung <nicolas.ph...@gmail.com> wrote: > Hello, > > @Akhil Das I'm trying to use the experimental API > https://github.com/apache/spark/blob/master/examples/scala-2.10/src/main/scala/org/apache/spark/examples/streaming/DirectKafkaWordCount.scala > <https://www.google.com/url?q=https%3A%2F%2Fgithub.com%2Fapache%2Fspark%2Fblob%2Fmaster%2Fexamples%2Fscala-2.10%2Fsrc%2Fmain%2Fscala%2Forg%2Fapache%2Fspark%2Fexamples%2Fstreaming%2FDirectKafkaWordCount.scala&sa=D&sntz=1&usg=AFQjCNFOmScaSfP-2J4Zn56k86-jHUkYaw>. > I'm reusing the same code snippet to initialize my topicSet. > > @Cody Koeninger I don't see any previous error messages (see the full log > at the end). To create the topic, I'm doing the following : > > "kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 > --partitions 10 --topic toto" > > "kafka-topics --create --zookeeper localhost:2181 --replication-factor 1 > --partitions 1 --topic toto-single" > > I'm launching my Spark Streaming in local mode. > > @Ted Yu There's no log "Couldn't connect to leader for topic", here's the > full version : > > "spark-submit --conf config.resource=application-integration.conf --class > nextgen.Main assembly-0.1-SNAPSHOT.jar > > 15/03/31 10:47:12 INFO SecurityManager: Changing view acls to: nphung > 15/03/31 10:47:12 INFO SecurityManager: Changing modify acls to: nphung > 15/03/31 10:47:12 INFO SecurityManager: SecurityManager: authentication > disabled; ui acls disabled; users with view permissions: Set(nphung); users > with modify permissions: Set(nphung) > 15/03/31 10:47:13 INFO Slf4jLogger: Slf4jLogger started > 15/03/31 10:47:13 INFO Remoting: Starting remoting > 15/03/31 10:47:13 INFO Remoting: Remoting started; listening on addresses > :[akka.tcp://sparkDriver@int.local:44180] > 15/03/31 10:47:13 INFO Remoting: Remoting now listens on addresses: > [akka.tcp://sparkDriver@int.local:44180] > 15/03/31 10:47:13 INFO Utils: Successfully started service 'sparkDriver' on > port 44180. > 15/03/31 10:47:13 INFO SparkEnv: Registering MapOutputTracker > 15/03/31 10:47:13 INFO SparkEnv: Registering BlockManagerMaster > 15/03/31 10:47:13 INFO DiskBlockManager: Created local directory at > /tmp/spark-local-20150331104713-2238 > 15/03/31 10:47:13 INFO MemoryStore: MemoryStore started with capacity 265.1 MB > 15/03/31 10:47:15 INFO HttpFileServer: HTTP File server directory is > /tmp/spark-2c8e34a0-bec3-4f1e-9fe7-83e08efc4f53 > 15/03/31 10:47:15 INFO HttpServer: Starting HTTP Server > 15/03/31 10:47:15 INFO Utils: Successfully started service 'HTTP file server' > on port 50204. > 15/03/31 10:47:15 INFO Utils: Successfully started service 'SparkUI' on port > 4040. > 15/03/31 10:47:15 INFO SparkUI: Started SparkUI at http://int.local:4040 > 15/03/31 10:47:16 INFO SparkContext: Added JAR > file:/home/nphung/assembly-0.1-SNAPSHOT.jar at > http://10.153.165.98:50204/jars/assembly-0.1-SNAPSHOT.jar with timestamp > 1427791636151 > 15/03/31 10:47:16 INFO AkkaUtils: Connecting to HeartbeatReceiver: > akka.tcp://sparkDriver@int.local:44180/user/HeartbeatReceiver > 15/03/31 10:47:16 INFO NettyBlockTransferService: Server created on 40630 > 15/03/31 10:47:16 INFO BlockManagerMaster: Trying to register BlockManager > 15/03/31 10:47:16 INFO BlockManagerMasterActor: Registering block manager > localhost:40630 with 265.1 MB RAM, BlockManagerId(<driver>, localhost, 40630) > 15/03/31 10:47:16 INFO BlockManagerMaster: Registered BlockManager > 15/03/31 10:47:17 INFO EventLoggingListener: Logging events to > hdfs://int.local:8020/user/spark/applicationHistory/local-1427791636195 > 15/03/31 10:47:17 INFO VerifiableProperties: Verifying properties > 15/03/31 10:47:17 INFO VerifiableProperties: Property group.id is overridden > to > 15/03/31 10:47:17 INFO VerifiableProperties: Property zookeeper.connect is > overridden to > 15/03/31 10:47:17 INFO ForEachDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO MappedDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO MappedDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Storage level = > StorageLevel(false, false, false, false, 1) > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Initialized and validated > org.apache.spark.streaming.kafka.DirectKafkaInputDStream@1daf3b44 > 15/03/31 10:47:17 INFO MappedDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO MappedDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO MappedDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Initialized and validated > org.apache.spark.streaming.dstream.MappedDStream@7fd8c559 > 15/03/31 10:47:17 INFO MappedDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO MappedDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO MappedDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Initialized and validated > org.apache.spark.streaming.dstream.MappedDStream@44c13103 > 15/03/31 10:47:17 INFO ForEachDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO ForEachDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO ForEachDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO ForEachDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO ForEachDStream: Initialized and validated > org.apache.spark.streaming.dstream.ForEachDStream@8f2098e > 15/03/31 10:47:17 INFO ForEachDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO MappedDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO MappedDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO MappedDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: metadataCleanupDelay = -1 > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Storage level = > StorageLevel(false, false, false, false, 1) > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO DirectKafkaInputDStream: Initialized and validated > org.apache.spark.streaming.kafka.DirectKafkaInputDStream@1daf3b44 > 15/03/31 10:47:17 INFO MappedDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO MappedDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO MappedDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Initialized and validated > org.apache.spark.streaming.dstream.MappedDStream@7fd8c559 > 15/03/31 10:47:17 INFO MappedDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO MappedDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO MappedDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Initialized and validated > org.apache.spark.streaming.dstream.MappedDStream@44c13103 > 15/03/31 10:47:17 INFO MappedDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO MappedDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO MappedDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO MappedDStream: Initialized and validated > org.apache.spark.streaming.dstream.MappedDStream@6c6366cf > 15/03/31 10:47:17 INFO ForEachDStream: Slide time = 2000 ms > 15/03/31 10:47:17 INFO ForEachDStream: Storage level = StorageLevel(false, > false, false, false, 1) > 15/03/31 10:47:17 INFO ForEachDStream: Checkpoint interval = null > 15/03/31 10:47:17 INFO ForEachDStream: Remember duration = 2000 ms > 15/03/31 10:47:17 INFO ForEachDStream: Initialized and validated > org.apache.spark.streaming.dstream.ForEachDStream@55a88417 > 15/03/31 10:47:17 INFO RecurringTimer: Started timer for JobGenerator at time > 1427791638000 > 15/03/31 10:47:17 INFO JobGenerator: Started JobGenerator at 1427791638000 ms > 15/03/31 10:47:17 INFO JobScheduler: Started JobScheduler > 15/03/31 10:47:18 INFO VerifiableProperties: Verifying properties > 15/03/31 10:47:18 INFO VerifiableProperties: Property group.id is overridden > to > 15/03/31 10:47:18 INFO VerifiableProperties: Property zookeeper.connect is > overridden to > 15/03/31 10:47:18 INFO JobScheduler: Added jobs for time 1427791638000 ms > 15/03/31 10:47:18 INFO JobScheduler: Starting job streaming job 1427791638000 > ms.0 from job set of time 1427791638000 ms > 15/03/31 10:47:18 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:18 INFO DAGScheduler: Got job 0 (print at Main.scala:59) with > 1 output partitions (allowLocal=true) > 15/03/31 10:47:18 INFO DAGScheduler: Final stage: Stage 0(print at > Main.scala:59) > 15/03/31 10:47:18 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:18 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:18 INFO DAGScheduler: Submitting Stage 0 (MappedRDD[2] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=0, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_0 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(1946) called with > curMem=3568, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_0_piece0 stored as bytes > in memory (estimated size 1946.0 B, free 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerInfo: Added broadcast_0_piece0 in memory > on localhost:40630 (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerMaster: Updated info of block > broadcast_0_piece0 > 15/03/31 10:47:18 INFO SparkContext: Created broadcast 0 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:18 INFO DAGScheduler: Submitting 1 missing tasks from Stage 0 > (MappedRDD[2] at map at Main.scala:50) > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Adding task set 0.0 with 1 tasks > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 0.0 in stage 0.0 (TID 0, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO Executor: Running task 0.0 in stage 0.0 (TID 0) > 15/03/31 10:47:18 INFO Executor: Fetching > http://10.153.165.98:50204/jars/assembly-0.1-SNAPSHOT.jar with timestamp > 1427791636151 > 15/03/31 10:47:18 INFO Utils: Fetching > http://10.153.165.98:50204/jars/assembly-0.1-SNAPSHOT.jar to > /tmp/fetchFileTemp4524847716079927807.tmp > 15/03/31 10:47:18 INFO Executor: Adding > file:/tmp/spark-d295d121-f359-4c7d-8725-e1f958ebff5a/assembly-0.1-SNAPSHOT.jar > to class loader > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 6 > 15/03/31 10:47:18 INFO Executor: Finished task 0.0 in stage 0.0 (TID 0). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO DAGScheduler: Stage 0 (print at Main.scala:59) > finished in 0,371 s > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 0.0 in stage 0.0 (TID 0) > in 350 ms on localhost (1/1) > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Removed TaskSet 0.0, whose tasks > have all completed, from pool > 15/03/31 10:47:18 INFO DAGScheduler: Job 0 finished: print at Main.scala:59, > took 0,717929 s > 15/03/31 10:47:18 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:18 INFO DAGScheduler: Got job 1 (print at Main.scala:59) with > 4 output partitions (allowLocal=true) > 15/03/31 10:47:18 INFO DAGScheduler: Final stage: Stage 1(print at > Main.scala:59) > 15/03/31 10:47:18 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:18 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:18 INFO DAGScheduler: Submitting Stage 1 (MappedRDD[2] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=5514, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_1 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(1946) called with > curMem=9082, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_1_piece0 stored as bytes > in memory (estimated size 1946.0 B, free 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerInfo: Added broadcast_1_piece0 in memory > on localhost:40630 (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerMaster: Updated info of block > broadcast_1_piece0 > 15/03/31 10:47:18 INFO SparkContext: Created broadcast 1 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:18 INFO DAGScheduler: Submitting 4 missing tasks from Stage 1 > (MappedRDD[2] at map at Main.scala:50) > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Adding task set 1.0 with 4 tasks > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 0.0 in stage 1.0 (TID 1, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 1.0 in stage 1.0 (TID 2, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 2.0 in stage 1.0 (TID 3, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 3.0 in stage 1.0 (TID 4, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO Executor: Running task 0.0 in stage 1.0 (TID 1) > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 3 > 15/03/31 10:47:18 INFO Executor: Running task 1.0 in stage 1.0 (TID 2) > 15/03/31 10:47:18 INFO Executor: Running task 2.0 in stage 1.0 (TID 3) > 15/03/31 10:47:18 INFO Executor: Running task 3.0 in stage 1.0 (TID 4) > 15/03/31 10:47:18 INFO Executor: Finished task 0.0 in stage 1.0 (TID 1). 570 > bytes result sent to driver > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 9 > 15/03/31 10:47:18 INFO Executor: Finished task 1.0 in stage 1.0 (TID 2). 570 > bytes result sent to driver > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 8 > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 0.0 in stage 1.0 (TID 1) > in 21 ms on localhost (1/4) > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 0 > 15/03/31 10:47:18 INFO Executor: Finished task 3.0 in stage 1.0 (TID 4). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO Executor: Finished task 2.0 in stage 1.0 (TID 3). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 1.0 in stage 1.0 (TID 2) > in 31 ms on localhost (2/4) > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 3.0 in stage 1.0 (TID 4) > in 39 ms on localhost (3/4) > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 2.0 in stage 1.0 (TID 3) > in 43 ms on localhost (4/4) > 15/03/31 10:47:18 INFO DAGScheduler: Stage 1 (print at Main.scala:59) > finished in 0,048 s > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Removed TaskSet 1.0, whose tasks > have all completed, from pool > 15/03/31 10:47:18 INFO DAGScheduler: Job 1 finished: print at Main.scala:59, > took 0,073960 s > 15/03/31 10:47:18 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:18 INFO DAGScheduler: Got job 2 (print at Main.scala:59) with > 5 output partitions (allowLocal=true) > 15/03/31 10:47:18 INFO DAGScheduler: Final stage: Stage 2(print at > Main.scala:59) > 15/03/31 10:47:18 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:18 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:18 INFO DAGScheduler: Submitting Stage 2 (MappedRDD[2] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=11028, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_2 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(1946) called with > curMem=14596, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_2_piece0 stored as bytes > in memory (estimated size 1946.0 B, free 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerInfo: Added broadcast_2_piece0 in memory > on localhost:40630 (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerMaster: Updated info of block > broadcast_2_piece0 > 15/03/31 10:47:18 INFO SparkContext: Created broadcast 2 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:18 INFO DAGScheduler: Submitting 5 missing tasks from Stage 2 > (MappedRDD[2] at map at Main.scala:50) > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Adding task set 2.0 with 5 tasks > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 0.0 in stage 2.0 (TID 5, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 1.0 in stage 2.0 (TID 6, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 2.0 in stage 2.0 (TID 7, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 3.0 in stage 2.0 (TID 8, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO Executor: Running task 0.0 in stage 2.0 (TID 5) > 15/03/31 10:47:18 INFO Executor: Running task 2.0 in stage 2.0 (TID 7) > 15/03/31 10:47:18 INFO Executor: Running task 1.0 in stage 2.0 (TID 6) > 15/03/31 10:47:18 INFO Executor: Running task 3.0 in stage 2.0 (TID 8) > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 2 > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 5 > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 7 > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 1 > 15/03/31 10:47:18 INFO Executor: Finished task 0.0 in stage 2.0 (TID 5). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO Executor: Finished task 1.0 in stage 2.0 (TID 6). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO TaskSetManager: Starting task 4.0 in stage 2.0 (TID 9, > localhost, ANY, 1266 bytes) > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 0.0 in stage 2.0 (TID 5) > in 12 ms on localhost (1/5) > 15/03/31 10:47:18 INFO Executor: Running task 4.0 in stage 2.0 (TID 9) > 15/03/31 10:47:18 INFO Executor: Finished task 3.0 in stage 2.0 (TID 8). 570 > bytes result sent to driver > 15/03/31 10:47:18 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 4 > 15/03/31 10:47:18 INFO Executor: Finished task 2.0 in stage 2.0 (TID 7). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 1.0 in stage 2.0 (TID 6) > in 18 ms on localhost (2/5) > 15/03/31 10:47:18 INFO Executor: Finished task 4.0 in stage 2.0 (TID 9). 570 > bytes result sent to driver > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 3.0 in stage 2.0 (TID 8) > in 23 ms on localhost (3/5) > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 2.0 in stage 2.0 (TID 7) > in 27 ms on localhost (4/5) > 15/03/31 10:47:18 INFO TaskSetManager: Finished task 4.0 in stage 2.0 (TID 9) > in 21 ms on localhost (5/5) > 15/03/31 10:47:18 INFO DAGScheduler: Stage 2 (print at Main.scala:59) > finished in 0,033 s > 15/03/31 10:47:18 INFO TaskSchedulerImpl: Removed TaskSet 2.0, whose tasks > have all completed, from pool > ------------------------------------------- > Time: 1427791638000 ms > ------------------------------------------- > 15/03/31 10:47:18 INFO DAGScheduler: Job 2 finished: print at Main.scala:59, > took 0,057381 s > > 15/03/31 10:47:18 INFO JobScheduler: Finished job streaming job 1427791638000 > ms.0 from job set of time 1427791638000 ms > 15/03/31 10:47:18 INFO JobScheduler: Starting job streaming job 1427791638000 > ms.1 from job set of time 1427791638000 ms > 15/03/31 10:47:18 INFO SparkContext: Starting job: foreachRDD at Main.scala:70 > 15/03/31 10:47:18 INFO DAGScheduler: Got job 3 (foreachRDD at Main.scala:70) > with 10 output partitions (allowLocal=false) > 15/03/31 10:47:18 INFO DAGScheduler: Final stage: Stage 3(foreachRDD at > Main.scala:70) > 15/03/31 10:47:18 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:18 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:18 INFO DAGScheduler: Submitting Stage 3 (MappedRDD[3] at map > at Main.scala:62), which has no missing parents > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(5064) called with > curMem=16542, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_3 stored as values in > memory (estimated size 4.9 KB, free 265.1 MB) > 15/03/31 10:47:18 INFO MemoryStore: ensureFreeSpace(2791) called with > curMem=21606, maxMem=278019440 > 15/03/31 10:47:18 INFO MemoryStore: Block broadcast_3_piece0 stored as bytes > in memory (estimated size 2.7 KB, free 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerInfo: Added broadcast_3_piece0 in memory > on localhost:40630 (size: 2.7 KB, free: 265.1 MB) > 15/03/31 10:47:18 INFO BlockManagerMaster: Updated info of block > broadcast_3_piece0 > 15/03/31 10:47:19 INFO SparkContext: Created broadcast 3 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:19 INFO DAGScheduler: Submitting 10 missing tasks from Stage 3 > (MappedRDD[3] at map at Main.scala:62) > 15/03/31 10:47:19 INFO TaskSchedulerImpl: Adding task set 3.0 with 10 tasks > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 0.0 in stage 3.0 (TID > 10, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 1.0 in stage 3.0 (TID > 11, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 2.0 in stage 3.0 (TID > 12, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 3.0 in stage 3.0 (TID > 13, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO Executor: Running task 0.0 in stage 3.0 (TID 10) > 15/03/31 10:47:19 INFO Executor: Running task 2.0 in stage 3.0 (TID 12) > 15/03/31 10:47:19 INFO Executor: Running task 1.0 in stage 3.0 (TID 11) > 15/03/31 10:47:19 INFO Executor: Running task 3.0 in stage 3.0 (TID 13) > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 6 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 3 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 9 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 0 > 15/03/31 10:47:19 INFO BlockManager: Removing broadcast 2 > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_2_piece0 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_2_piece0 of size 1946 > dropped from memory (free 277996989) > 15/03/31 10:47:19 INFO BlockManagerInfo: Removed broadcast_2_piece0 on > localhost:40630 in memory (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:19 INFO BlockManagerMaster: Updated info of block > broadcast_2_piece0 > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_2 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_2 of size 3568 dropped > from memory (free 278000557) > 15/03/31 10:47:19 INFO ContextCleaner: Cleaned broadcast 2 > 15/03/31 10:47:19 INFO BlockManager: Removing broadcast 1 > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_1 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_1 of size 3568 dropped > from memory (free 278004125) > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_1_piece0 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_1_piece0 of size 1946 > dropped from memory (free 278006071) > 15/03/31 10:47:19 INFO BlockManagerInfo: Removed broadcast_1_piece0 on > localhost:40630 in memory (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:19 INFO BlockManagerMaster: Updated info of block > broadcast_1_piece0 > 15/03/31 10:47:19 INFO ContextCleaner: Cleaned broadcast 1 > 15/03/31 10:47:19 INFO BlockManager: Removing broadcast 0 > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_0 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_0 of size 3568 dropped > from memory (free 278009639) > 15/03/31 10:47:19 INFO BlockManager: Removing block broadcast_0_piece0 > 15/03/31 10:47:19 INFO MemoryStore: Block broadcast_0_piece0 of size 1946 > dropped from memory (free 278011585) > 15/03/31 10:47:19 INFO BlockManagerInfo: Removed broadcast_0_piece0 on > localhost:40630 in memory (size: 1946.0 B, free: 265.1 MB) > 15/03/31 10:47:19 INFO BlockManagerMaster: Updated info of block > broadcast_0_piece0 > 15/03/31 10:47:19 INFO ContextCleaner: Cleaned broadcast 0 > 15/03/31 10:47:19 INFO Executor: Finished task 3.0 in stage 3.0 (TID 13). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO Executor: Finished task 0.0 in stage 3.0 (TID 10). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 4.0 in stage 3.0 (TID > 14, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO Executor: Running task 4.0 in stage 3.0 (TID 14) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 3.0 in stage 3.0 (TID > 13) in 303 ms on localhost (1/10) > 15/03/31 10:47:19 INFO Executor: Finished task 1.0 in stage 3.0 (TID 11). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO Executor: Finished task 2.0 in stage 3.0 (TID 12). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 5.0 in stage 3.0 (TID > 15, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 8 > 15/03/31 10:47:19 INFO Executor: Running task 5.0 in stage 3.0 (TID 15) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 0.0 in stage 3.0 (TID > 10) in 312 ms on localhost (2/10) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 1.0 in stage 3.0 (TID > 11) in 314 ms on localhost (3/10) > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 6.0 in stage 3.0 (TID > 16, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 7.0 in stage 3.0 (TID > 17, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO Executor: Running task 6.0 in stage 3.0 (TID 16) > 15/03/31 10:47:19 INFO Executor: Running task 7.0 in stage 3.0 (TID 17) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 2.0 in stage 3.0 (TID > 12) in 317 ms on localhost (4/10) > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 5 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 1 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 2 > 15/03/31 10:47:19 INFO Executor: Finished task 5.0 in stage 3.0 (TID 15). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 8.0 in stage 3.0 (TID > 18, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO Executor: Finished task 7.0 in stage 3.0 (TID 17). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO TaskSetManager: Starting task 9.0 in stage 3.0 (TID > 19, localhost, ANY, 1266 bytes) > 15/03/31 10:47:19 INFO Executor: Finished task 4.0 in stage 3.0 (TID 14). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO Executor: Running task 9.0 in stage 3.0 (TID 19) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 7.0 in stage 3.0 (TID > 17) in 40 ms on localhost (5/10) > 15/03/31 10:47:19 INFO Executor: Finished task 6.0 in stage 3.0 (TID 16). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO Executor: Running task 8.0 in stage 3.0 (TID 18) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 5.0 in stage 3.0 (TID > 15) in 53 ms on localhost (6/10) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 4.0 in stage 3.0 (TID > 14) in 59 ms on localhost (7/10) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 6.0 in stage 3.0 (TID > 16) in 50 ms on localhost (8/10) > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 4 > 15/03/31 10:47:19 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 7 > 15/03/31 10:47:19 INFO Executor: Finished task 8.0 in stage 3.0 (TID 18). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO Executor: Finished task 9.0 in stage 3.0 (TID 19). 569 > bytes result sent to driver > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 8.0 in stage 3.0 (TID > 18) in 49 ms on localhost (9/10) > 15/03/31 10:47:19 INFO TaskSetManager: Finished task 9.0 in stage 3.0 (TID > 19) in 50 ms on localhost (10/10) > 15/03/31 10:47:19 INFO TaskSchedulerImpl: Removed TaskSet 3.0, whose tasks > have all completed, from pool > 15/03/31 10:47:19 INFO DAGScheduler: Stage 3 (foreachRDD at Main.scala:70) > finished in 0,410 s > 15/03/31 10:47:19 INFO DAGScheduler: Job 3 finished: foreachRDD at > Main.scala:70, took 0,438664 s > 15/03/31 10:47:19 INFO JobScheduler: Finished job streaming job 1427791638000 > ms.1 from job set of time 1427791638000 ms > 15/03/31 10:47:19 INFO JobScheduler: Total delay: 1,420 s for time > 1427791638000 ms (execution: 1,372 s) > 15/03/31 10:47:19 INFO ReceivedBlockTracker: Deleting batches ArrayBuffer() > 15/03/31 10:47:20 INFO JobScheduler: Starting job streaming job 1427791640000 > ms.0 from job set of time 1427791640000 ms > 15/03/31 10:47:20 INFO JobScheduler: Added jobs for time 1427791640000 ms > 15/03/31 10:47:20 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:20 INFO DAGScheduler: Got job 4 (print at Main.scala:59) with > 1 output partitions (allowLocal=true) > 15/03/31 10:47:20 INFO DAGScheduler: Final stage: Stage 4(print at > Main.scala:59) > 15/03/31 10:47:20 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:20 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:20 INFO DAGScheduler: Submitting Stage 4 (MappedRDD[6] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=7855, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_4 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(1945) called with > curMem=11423, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_4_piece0 stored as bytes > in memory (estimated size 1945.0 B, free 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerInfo: Added broadcast_4_piece0 in memory > on localhost:40630 (size: 1945.0 B, free: 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerMaster: Updated info of block > broadcast_4_piece0 > 15/03/31 10:47:20 INFO SparkContext: Created broadcast 4 from getCallSite at > DStream.scala:294 > 15/03/31 10:47:20 INFO DAGScheduler: Submitting 1 missing tasks from Stage 4 > (MappedRDD[6] at map at Main.scala:50) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Adding task set 4.0 with 1 tasks > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 0.0 in stage 4.0 (TID > 20, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 0.0 in stage 4.0 (TID 20) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 6 > 15/03/31 10:47:20 INFO Executor: Finished task 0.0 in stage 4.0 (TID 20). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 0.0 in stage 4.0 (TID > 20) in 6 ms on localhost (1/1) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Removed TaskSet 4.0, whose tasks > have all completed, from pool > 15/03/31 10:47:20 INFO DAGScheduler: Stage 4 (print at Main.scala:59) > finished in 0,007 s > 15/03/31 10:47:20 INFO DAGScheduler: Job 4 finished: print at Main.scala:59, > took 0,021293 s > 15/03/31 10:47:20 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:20 INFO DAGScheduler: Got job 5 (print at Main.scala:59) with > 4 output partitions (allowLocal=true) > 15/03/31 10:47:20 INFO DAGScheduler: Final stage: Stage 5(print at > Main.scala:59) > 15/03/31 10:47:20 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:20 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:20 INFO DAGScheduler: Submitting Stage 5 (MappedRDD[6] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=13368, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_5 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(1945) called with > curMem=16936, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_5_piece0 stored as bytes > in memory (estimated size 1945.0 B, free 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerInfo: Added broadcast_5_piece0 in memory > on localhost:40630 (size: 1945.0 B, free: 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerMaster: Updated info of block > broadcast_5_piece0 > 15/03/31 10:47:20 INFO SparkContext: Created broadcast 5 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:20 INFO DAGScheduler: Submitting 4 missing tasks from Stage 5 > (MappedRDD[6] at map at Main.scala:50) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Adding task set 5.0 with 4 tasks > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 0.0 in stage 5.0 (TID > 21, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 1.0 in stage 5.0 (TID > 22, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 2.0 in stage 5.0 (TID > 23, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 3.0 in stage 5.0 (TID > 24, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 0.0 in stage 5.0 (TID 21) > 15/03/31 10:47:20 INFO Executor: Running task 2.0 in stage 5.0 (TID 23) > 15/03/31 10:47:20 INFO Executor: Running task 3.0 in stage 5.0 (TID 24) > 15/03/31 10:47:20 INFO Executor: Running task 1.0 in stage 5.0 (TID 22) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 0 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 8 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 9 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 3 > 15/03/31 10:47:20 INFO Executor: Finished task 2.0 in stage 5.0 (TID 23). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Finished task 3.0 in stage 5.0 (TID 24). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Finished task 0.0 in stage 5.0 (TID 21). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 2.0 in stage 5.0 (TID > 23) in 7 ms on localhost (1/4) > 15/03/31 10:47:20 INFO Executor: Finished task 1.0 in stage 5.0 (TID 22). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 3.0 in stage 5.0 (TID > 24) in 10 ms on localhost (2/4) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 0.0 in stage 5.0 (TID > 21) in 14 ms on localhost (3/4) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 1.0 in stage 5.0 (TID > 22) in 16 ms on localhost (4/4) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Removed TaskSet 5.0, whose tasks > have all completed, from pool > 15/03/31 10:47:20 INFO DAGScheduler: Stage 5 (print at Main.scala:59) > finished in 0,020 s > 15/03/31 10:47:20 INFO DAGScheduler: Job 5 finished: print at Main.scala:59, > took 0,034514 s > 15/03/31 10:47:20 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:20 INFO DAGScheduler: Got job 6 (print at Main.scala:59) with > 5 output partitions (allowLocal=true) > 15/03/31 10:47:20 INFO DAGScheduler: Final stage: Stage 6(print at > Main.scala:59) > 15/03/31 10:47:20 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:20 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:20 INFO DAGScheduler: Submitting Stage 6 (MappedRDD[6] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=18881, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_6 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(1945) called with > curMem=22449, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_6_piece0 stored as bytes > in memory (estimated size 1945.0 B, free 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerInfo: Added broadcast_6_piece0 in memory > on localhost:40630 (size: 1945.0 B, free: 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerMaster: Updated info of block > broadcast_6_piece0 > 15/03/31 10:47:20 INFO SparkContext: Created broadcast 6 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:20 INFO DAGScheduler: Submitting 5 missing tasks from Stage 6 > (MappedRDD[6] at map at Main.scala:50) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Adding task set 6.0 with 5 tasks > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 0.0 in stage 6.0 (TID > 25, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 1.0 in stage 6.0 (TID > 26, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 2.0 in stage 6.0 (TID > 27, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 3.0 in stage 6.0 (TID > 28, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 0.0 in stage 6.0 (TID 25) > 15/03/31 10:47:20 INFO Executor: Running task 2.0 in stage 6.0 (TID 27) > 15/03/31 10:47:20 INFO Executor: Running task 3.0 in stage 6.0 (TID 28) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 2 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 1 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 7 > 15/03/31 10:47:20 INFO Executor: Finished task 0.0 in stage 6.0 (TID 25). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Running task 1.0 in stage 6.0 (TID 26) > 15/03/31 10:47:20 INFO Executor: Finished task 3.0 in stage 6.0 (TID 28). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 4.0 in stage 6.0 (TID > 29, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 5 > 15/03/31 10:47:20 INFO Executor: Finished task 2.0 in stage 6.0 (TID 27). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 0.0 in stage 6.0 (TID > 25) in 8 ms on localhost (1/5) > 15/03/31 10:47:20 INFO Executor: Finished task 1.0 in stage 6.0 (TID 26). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Running task 4.0 in stage 6.0 (TID 29) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 4 > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 3.0 in stage 6.0 (TID > 28) in 10 ms on localhost (2/5) > 15/03/31 10:47:20 INFO Executor: Finished task 4.0 in stage 6.0 (TID 29). 570 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 2.0 in stage 6.0 (TID > 27) in 13 ms on localhost (3/5) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 1.0 in stage 6.0 (TID > 26) in 15 ms on localhost (4/5) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 4.0 in stage 6.0 (TID > 29) in 12 ms on localhost (5/5) > 15/03/31 10:47:20 INFO DAGScheduler: Stage 6 (print at Main.scala:59) > finished in 0,019 s > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Removed TaskSet 6.0, whose tasks > have all completed, from pool > ------------------------------------------- > Time: 1427791640000 ms > ------------------------------------------- > > 15/03/31 10:47:20 INFO DAGScheduler: Job 6 finished: print at Main.scala:59, > took 0,034922 s > 15/03/31 10:47:20 INFO JobScheduler: Finished job streaming job 1427791640000 > ms.0 from job set of time 1427791640000 ms > 15/03/31 10:47:20 INFO SparkContext: Starting job: foreachRDD at Main.scala:70 > 15/03/31 10:47:20 INFO DAGScheduler: Got job 7 (foreachRDD at Main.scala:70) > with 10 output partitions (allowLocal=false) > 15/03/31 10:47:20 INFO DAGScheduler: Final stage: Stage 7(foreachRDD at > Main.scala:70) > 15/03/31 10:47:20 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:20 INFO JobScheduler: Starting job streaming job 1427791640000 > ms.1 from job set of time 1427791640000 ms > 15/03/31 10:47:20 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:20 INFO DAGScheduler: Submitting Stage 7 (MappedRDD[7] at map > at Main.scala:62), which has no missing parents > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(5064) called with > curMem=24394, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_7 stored as values in > memory (estimated size 4.9 KB, free 265.1 MB) > 15/03/31 10:47:20 INFO MemoryStore: ensureFreeSpace(2792) called with > curMem=29458, maxMem=278019440 > 15/03/31 10:47:20 INFO MemoryStore: Block broadcast_7_piece0 stored as bytes > in memory (estimated size 2.7 KB, free 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerInfo: Added broadcast_7_piece0 in memory > on localhost:40630 (size: 2.7 KB, free: 265.1 MB) > 15/03/31 10:47:20 INFO BlockManagerMaster: Updated info of block > broadcast_7_piece0 > 15/03/31 10:47:20 INFO SparkContext: Created broadcast 7 from getCallSite at > DStream.scala:294 > 15/03/31 10:47:20 INFO DAGScheduler: Submitting 10 missing tasks from Stage 7 > (MappedRDD[7] at map at Main.scala:62) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Adding task set 7.0 with 10 tasks > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 0.0 in stage 7.0 (TID > 30, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 1.0 in stage 7.0 (TID > 31, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 2.0 in stage 7.0 (TID > 32, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 3.0 in stage 7.0 (TID > 33, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 0.0 in stage 7.0 (TID 30) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 6 > 15/03/31 10:47:20 INFO Executor: Running task 1.0 in stage 7.0 (TID 31) > 15/03/31 10:47:20 INFO Executor: Running task 2.0 in stage 7.0 (TID 32) > 15/03/31 10:47:20 INFO Executor: Running task 3.0 in stage 7.0 (TID 33) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 9 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 0 > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 3 > 15/03/31 10:47:20 INFO Executor: Finished task 0.0 in stage 7.0 (TID 30). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Finished task 2.0 in stage 7.0 (TID 32). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Finished task 1.0 in stage 7.0 (TID 31). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 4.0 in stage 7.0 (TID > 34, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 4.0 in stage 7.0 (TID 34) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 5.0 in stage 7.0 (TID > 35, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 6.0 in stage 7.0 (TID > 36, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 8 > 15/03/31 10:47:20 INFO Executor: Running task 6.0 in stage 7.0 (TID 36) > 15/03/31 10:47:20 INFO Executor: Finished task 3.0 in stage 7.0 (TID 33). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO Executor: Running task 5.0 in stage 7.0 (TID 35) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 0.0 in stage 7.0 (TID > 30) in 34 ms on localhost (1/10) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 5 > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 1.0 in stage 7.0 (TID > 31) in 42 ms on localhost (2/10) > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 7.0 in stage 7.0 (TID > 37, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 2 > 15/03/31 10:47:20 INFO Executor: Running task 7.0 in stage 7.0 (TID 37) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 2.0 in stage 7.0 (TID > 32) in 45 ms on localhost (3/10) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 1 > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 3.0 in stage 7.0 (TID > 33) in 46 ms on localhost (4/10) > 15/03/31 10:47:20 INFO Executor: Finished task 4.0 in stage 7.0 (TID 34). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 8.0 in stage 7.0 (TID > 38, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO Executor: Running task 8.0 in stage 7.0 (TID 38) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 4.0 in stage 7.0 (TID > 34) in 31 ms on localhost (5/10) > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 7 > 15/03/31 10:47:20 INFO Executor: Finished task 6.0 in stage 7.0 (TID 36). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Starting task 9.0 in stage 7.0 (TID > 39, localhost, ANY, 1266 bytes) > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 6.0 in stage 7.0 (TID > 36) in 35 ms on localhost (6/10) > 15/03/31 10:47:20 INFO Executor: Running task 9.0 in stage 7.0 (TID 39) > 15/03/31 10:47:20 INFO Executor: Finished task 5.0 in stage 7.0 (TID 35). 569 > bytes result sent to driver > 15/03/31 10:47:20 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 4 > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 5.0 in stage 7.0 (TID > 35) in 39 ms on localhost (7/10) > 15/03/31 10:47:20 INFO Executor: Finished task 7.0 in stage 7.0 (TID 37). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 7.0 in stage 7.0 (TID > 37) in 36 ms on localhost (8/10) > 15/03/31 10:47:20 INFO Executor: Finished task 8.0 in stage 7.0 (TID 38). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 8.0 in stage 7.0 (TID > 38) in 26 ms on localhost (9/10) > 15/03/31 10:47:20 INFO Executor: Finished task 9.0 in stage 7.0 (TID 39). 569 > bytes result sent to driver > 15/03/31 10:47:20 INFO TaskSetManager: Finished task 9.0 in stage 7.0 (TID > 39) in 27 ms on localhost (10/10) > 15/03/31 10:47:20 INFO TaskSchedulerImpl: Removed TaskSet 7.0, whose tasks > have all completed, from pool > 15/03/31 10:47:20 INFO DAGScheduler: Stage 7 (foreachRDD at Main.scala:70) > finished in 0,098 s > 15/03/31 10:47:20 INFO DAGScheduler: Job 7 finished: foreachRDD at > Main.scala:70, took 0,111899 s > 15/03/31 10:47:20 INFO JobScheduler: Finished job streaming job 1427791640000 > ms.1 from job set of time 1427791640000 ms > 15/03/31 10:47:20 INFO JobScheduler: Total delay: 0,231 s for time > 1427791640000 ms (execution: 0,213 s) > 15/03/31 10:47:20 INFO MappedRDD: Removing RDD 2 from persistence list > 15/03/31 10:47:20 INFO BlockManager: Removing RDD 2 > 15/03/31 10:47:20 INFO MappedRDD: Removing RDD 1 from persistence list > 15/03/31 10:47:20 INFO KafkaRDD: Removing RDD 0 from persistence list > 15/03/31 10:47:20 INFO BlockManager: Removing RDD 0 > 15/03/31 10:47:20 INFO MappedRDD: Removing RDD 3 from persistence list > 15/03/31 10:47:20 INFO BlockManager: Removing RDD 3 > 15/03/31 10:47:20 INFO BlockManager: Removing RDD 1 > 15/03/31 10:47:20 INFO ReceivedBlockTracker: Deleting batches ArrayBuffer() > 15/03/31 10:47:22 INFO JobScheduler: Added jobs for time 1427791642000 ms > 15/03/31 10:47:22 INFO JobScheduler: Starting job streaming job 1427791642000 > ms.0 from job set of time 1427791642000 ms > 15/03/31 10:47:22 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:22 INFO DAGScheduler: Got job 8 (print at Main.scala:59) with > 1 output partitions (allowLocal=true) > 15/03/31 10:47:22 INFO DAGScheduler: Final stage: Stage 8(print at > Main.scala:59) > 15/03/31 10:47:22 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:22 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:22 INFO DAGScheduler: Submitting Stage 8 (MappedRDD[10] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:22 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=32250, maxMem=278019440 > 15/03/31 10:47:22 INFO MemoryStore: Block broadcast_8 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:22 INFO MemoryStore: ensureFreeSpace(1949) called with > curMem=35818, maxMem=278019440 > 15/03/31 10:47:22 INFO MemoryStore: Block broadcast_8_piece0 stored as bytes > in memory (estimated size 1949.0 B, free 265.1 MB) > 15/03/31 10:47:22 INFO BlockManagerInfo: Added broadcast_8_piece0 in memory > on localhost:40630 (size: 1949.0 B, free: 265.1 MB) > 15/03/31 10:47:22 INFO BlockManagerMaster: Updated info of block > broadcast_8_piece0 > 15/03/31 10:47:22 INFO SparkContext: Created broadcast 8 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:22 INFO DAGScheduler: Submitting 1 missing tasks from Stage 8 > (MappedRDD[10] at map at Main.scala:50) > 15/03/31 10:47:22 INFO TaskSchedulerImpl: Adding task set 8.0 with 1 tasks > 15/03/31 10:47:22 INFO TaskSetManager: Starting task 0.0 in stage 8.0 (TID > 40, localhost, ANY, 1266 bytes) > 15/03/31 10:47:22 INFO Executor: Running task 0.0 in stage 8.0 (TID 40) > 15/03/31 10:47:22 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 6 > 15/03/31 10:47:22 INFO Executor: Finished task 0.0 in stage 8.0 (TID 40). 570 > bytes result sent to driver > 15/03/31 10:47:22 INFO TaskSetManager: Finished task 0.0 in stage 8.0 (TID > 40) in 6 ms on localhost (1/1) > 15/03/31 10:47:22 INFO TaskSchedulerImpl: Removed TaskSet 8.0, whose tasks > have all completed, from pool > 15/03/31 10:47:22 INFO DAGScheduler: Stage 8 (print at Main.scala:59) > finished in 0,008 s > 15/03/31 10:47:22 INFO DAGScheduler: Job 8 finished: print at Main.scala:59, > took 0,021082 s > 15/03/31 10:47:22 INFO SparkContext: Starting job: print at Main.scala:59 > 15/03/31 10:47:22 INFO DAGScheduler: Got job 9 (print at Main.scala:59) with > 4 output partitions (allowLocal=true) > 15/03/31 10:47:22 INFO DAGScheduler: Final stage: Stage 9(print at > Main.scala:59) > 15/03/31 10:47:22 INFO DAGScheduler: Parents of final stage: List() > 15/03/31 10:47:22 INFO DAGScheduler: Missing parents: List() > 15/03/31 10:47:22 INFO DAGScheduler: Submitting Stage 9 (MappedRDD[10] at map > at Main.scala:50), which has no missing parents > 15/03/31 10:47:22 INFO MemoryStore: ensureFreeSpace(3568) called with > curMem=37767, maxMem=278019440 > 15/03/31 10:47:22 INFO MemoryStore: Block broadcast_9 stored as values in > memory (estimated size 3.5 KB, free 265.1 MB) > 15/03/31 10:47:22 INFO MemoryStore: ensureFreeSpace(1949) called with > curMem=41335, maxMem=278019440 > 15/03/31 10:47:22 INFO MemoryStore: Block broadcast_9_piece0 stored as bytes > in memory (estimated size 1949.0 B, free 265.1 MB) > 15/03/31 10:47:22 INFO BlockManagerInfo: Added broadcast_9_piece0 in memory > on localhost:40630 (size: 1949.0 B, free: 265.1 MB) > 15/03/31 10:47:22 INFO BlockManagerMaster: Updated info of block > broadcast_9_piece0 > 15/03/31 10:47:22 INFO SparkContext: Created broadcast 9 from broadcast at > DAGScheduler.scala:838 > 15/03/31 10:47:22 INFO DAGScheduler: Submitting 4 missing tasks from Stage 9 > (MappedRDD[10] at map at Main.scala:50) > 15/03/31 10:47:22 INFO TaskSchedulerImpl: Adding task set 9.0 with 4 tasks > 15/03/31 10:47:22 INFO TaskSetManager: Starting task 0.0 in stage 9.0 (TID > 41, localhost, ANY, 1266 bytes) > 15/03/31 10:47:22 INFO TaskSetManager: Starting task 1.0 in stage 9.0 (TID > 42, localhost, ANY, 1266 bytes) > 15/03/31 10:47:22 INFO TaskSetManager: Starting task 2.0 in stage 9.0 (TID > 43, localhost, ANY, 1266 bytes) > 15/03/31 10:47:22 INFO TaskSetManager: Starting task 3.0 in stage 9.0 (TID > 44, localhost, ANY, 1266 bytes) > 15/03/31 10:47:22 INFO Executor: Running task 2.0 in stage 9.0 (TID 43) > 15/03/31 10:47:22 INFO Executor: Running task 1.0 in stage 9.0 (TID 42) > 15/03/31 10:47:22 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 0 > 15/03/31 10:47:22 INFO Executor: Running task 0.0 in stage 9.0 (TID 41) > 15/03/31 10:47:22 WARN KafkaRDD: Beginning offset ${part.fromOffset} is the > same as ending offset skipping toto 3 > 15/03/31 10:47:22 INFO Executor: Finished task 2.0 in stage 9.0 (TID 43). 570 > bytes result sent to driver > 15/03/31 10:47:22 INFO Executor: Finished task 0.0 in stage 9.0 (TID 41). 570 > bytes result sent to driver > 15/03/31 10:47:22 INFO Executor: Running task 3.0 in stage 9.0 (TID 44) > 15/03/31 10:47:22 INFO TaskSetManager: Finished task 2.0 in stage 9.0 (TID > 43) in 5 ms on localhost (1/4) > 15/03/31 10:47:22 INFO KafkaRDD: Computing topic toto, partition 9 offsets 0 > -> 1 > 15/03/31 10:47:22 INFO TaskSetManager: Finished task 0.0 in stage 9.0 (TID > 41) in 9 ms on localhost (2/4) > 15/03/31 10:47:22 INFO VerifiableProperties: Verifying properties > 15/03/31 10:47:22 INFO VerifiableProperties: Property group.id is overridden > to > 15/03/31 10:47:22 INFO VerifiableProperties: Property zookeeper.connect is > overridden to > 15/03/31 10:47:22 INFO KafkaRDD: Computing topic toto, partition 8 offsets 0 > -> 1 > 15/03/31 10:47:22 ERROR TaskContextImpl: Error in TaskCompletionListener > java.lang.NullPointerException > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.close(KafkaRDD.scala:158) > at org.apache.spark.util.NextIterator.closeIfNeeded(NextIterator.scala:63) > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) > at > org.apache.spark.TaskContextImpl$$anon$1.onTaskCompletion(TaskContextImpl.scala:49) > at > org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:68) > at > org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:66) > at > scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) > at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) > at > org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:66) > at org.apache.spark.scheduler.Task.run(Task.scala:58) > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) > 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) > 15/03/31 10:47:22 INFO VerifiableProperties: Verifying properties > 15/03/31 10:47:22 INFO VerifiableProperties: Property group.id is overridden > to > 15/03/31 10:47:22 INFO VerifiableProperties: Property zookeeper.connect is > overridden to > 15/03/31 10:47:22 ERROR TaskContextImpl: Error in TaskCompletionListener > java.lang.NullPointerException > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.close(KafkaRDD.scala:158) > at org.apache.spark.util.NextIterator.closeIfNeeded(NextIterator.scala:63) > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) > at > org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) > at > org.apache.spark.TaskContextImpl$$anon$1.onTaskCompletion(TaskContextImpl.scala:49) > at > org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:68) > at > org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:66) > at > scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) > at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) > at > org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:66) > at org.apache.spark.scheduler.Task.run(Task.scala:58) > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) > 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) > 15/03/31 10:47:22 ERROR Executor: Exception in task 1.0 in stage 9.0 (TID 42) > org.apache.spark.util.TaskCompletionListenerException > at > org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) > at org.apache.spark.scheduler.Task.run(Task.scala:58) > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) > 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) > 15/03/31 10:47:22 ERROR Executor: Exception in task 3.0 in stage 9.0 (TID 44) > org.apache.spark.util.TaskCompletionListenerException > at > org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) > at org.apache.spark.scheduler.Task.run(Task.scala:58) > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) > 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) > 15/03/31 10:47:22 WARN TaskSetManager: Lost task 3.0 in stage 9.0 (TID 44, > localhost): org.apache.spark.util.TaskCompletionListenerException > at > org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) > at org.apache.spark.scheduler.Task.run(Task.scala:58) > at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) > 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) > > 15/03/31 10:47:22 ERROR TaskSetManager: Task 3 in stage 9.0 failed 1 > times; aborting job" > > I don't know if this could help but using the Spark UI. I find some > additional log when the tasks failed : > > org.apache.spark.streaming.dstream.DStream.print(DStream.scala:624) > > which corresponds to > https://github.com/cloudera/spark/blob/cdh5.3.2-release/streaming/src/main/scala/org/apache/spark/streaming/dstream/DStream.scala > > For integration, I'm using the Spark version ship with cloudera 5.3.2 > whereas in development Spark 1.3.0. Can this cause some troubles ? > > Thanks for your answer. > Regards, > Nicolas PHUNG > > On Mon, Mar 30, 2015 at 7:05 PM, Ted Yu <yuzhih...@gmail.com> wrote: > >> Nicolas: >> See if there was occurrence of the following exception in the log: >> errs => throw new SparkException( >> s"Couldn't connect to leader for topic ${part.topic} >> ${part.partition}: " + >> errs.mkString("\n")), >> >> Cheers >> >> On Mon, Mar 30, 2015 at 9:40 AM, Cody Koeninger <c...@koeninger.org> >> wrote: >> >>> This line >>> >>> at org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.close( >>> KafkaRDD.scala:158) >>> >>> is the attempt to close the underlying kafka simple consumer. >>> >>> We can add a null pointer check, but the underlying issue of the >>> consumer being null probably indicates a problem earlier. Do you see any >>> previous error messages? >>> >>> Also, can you clarify for the successful and failed cases which topics >>> you are attempting this on, how many partitions there are, and whether >>> there are messages in the partitions? There's an existing jira regarding >>> empty partitions. >>> >>> >>> >>> >>> On Mon, Mar 30, 2015 at 11:05 AM, Nicolas Phung <nicolas.ph...@gmail.com >>> > wrote: >>> >>>> Hello, >>>> >>>> I'm using spark-streaming-kafka 1.3.0 with the new consumer "Approach >>>> 2: Direct Approach (No Receivers)" ( >>>> http://spark.apache.org/docs/latest/streaming-kafka-integration.html). >>>> I'm using the following code snippets : >>>> >>>> // Create direct kafka stream with brokers and topics >>>> val messages = KafkaUtils.createDirectStream[String, Array[Byte], >>>> StringDecoder, DefaultDecoder]( >>>> ssc, kafkaParams, topicsSet) >>>> >>>> // Get the stuff from Kafka and print them >>>> val raw = messages.map(_._2) >>>> val dStream: DStream[RawScala] = raw.map( >>>> byte => { >>>> // Avro Decoder >>>> println("Byte length: " + byte.length) >>>> val rawDecoder = new AvroDecoder[Raw](schema = Raw.getClassSchema) >>>> RawScala.toScala(rawDecoder.fromBytes(byte)) >>>> } >>>> ) >>>> // Useful for debug >>>> dStream.print() >>>> >>>> I launch my Spark Streaming and everything is fine if there's no >>>> incoming logs from Kafka. When I'm sending a log, I got the following error >>>> : >>>> >>>> 15/03/30 17:34:40 ERROR TaskContextImpl: Error in TaskCompletionListener >>>> java.lang.NullPointerException >>>> at >>>> org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator.close(KafkaRDD.scala:158) >>>> at >>>> org.apache.spark.util.NextIterator.closeIfNeeded(NextIterator.scala:63) >>>> at >>>> org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) >>>> at >>>> org.apache.spark.streaming.kafka.KafkaRDD$KafkaRDDIterator$$anonfun$1.apply(KafkaRDD.scala:101) >>>> at >>>> org.apache.spark.TaskContextImpl$$anon$1.onTaskCompletion(TaskContextImpl.scala:49) >>>> at >>>> org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:68) >>>> at >>>> org.apache.spark.TaskContextImpl$$anonfun$markTaskCompleted$1.apply(TaskContextImpl.scala:66) >>>> at >>>> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) >>>> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) >>>> at >>>> org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:66) >>>> at org.apache.spark.scheduler.Task.run(Task.scala:58) >>>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) >>>> 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) >>>> 15/03/30 17:34:40 INFO TaskSetManager: Finished task 3.0 in stage 28.0 >>>> (TID 94) in 12 ms on localhost (2/4) >>>> 15/03/30 17:34:40 INFO TaskSetManager: Finished task 2.0 in stage 28.0 >>>> (TID 93) in 13 ms on localhost (3/4) >>>> 15/03/30 17:34:40 ERROR Executor: Exception in task 0.0 in stage 28.0 >>>> (TID 91) >>>> org.apache.spark.util.TaskCompletionListenerException >>>> at >>>> org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) >>>> at org.apache.spark.scheduler.Task.run(Task.scala:58) >>>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) >>>> 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) >>>> 15/03/30 17:34:40 WARN TaskSetManager: Lost task 0.0 in stage 28.0 (TID >>>> 91, localhost): org.apache.spark.util.TaskCompletionListenerException >>>> at >>>> org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) >>>> at org.apache.spark.scheduler.Task.run(Task.scala:58) >>>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) >>>> 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) >>>> >>>> 15/03/30 17:34:40 ERROR TaskSetManager: Task 0 in stage 28.0 failed 1 >>>> times; aborting job >>>> 15/03/30 17:34:40 INFO TaskSchedulerImpl: Removed TaskSet 28.0, whose >>>> tasks have all completed, from pool >>>> 15/03/30 17:34:40 INFO TaskSchedulerImpl: Cancelling stage 28 >>>> 15/03/30 17:34:40 INFO DAGScheduler: Job 28 failed: print at >>>> HotFCANextGen.scala:63, took 0,041068 s >>>> 15/03/30 17:34:40 INFO JobScheduler: Starting job streaming job >>>> 1427729680000 ms.1 from job set of time 1427729680000 ms >>>> 15/03/30 17:34:40 ERROR JobScheduler: Error running job streaming job >>>> 1427729680000 ms.0 >>>> org.apache.spark.SparkException: Job aborted due to stage failure: Task >>>> 0 in stage 28.0 failed 1 times, most recent failure: Lost task 0.0 in stage >>>> 28.0 (TID 91, localhost): >>>> org.apache.spark.util.TaskCompletionListenerException >>>> at >>>> org.apache.spark.TaskContextImpl.markTaskCompleted(TaskContextImpl.scala:76) >>>> at org.apache.spark.scheduler.Task.run(Task.scala:58) >>>> at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:196) >>>> 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) >>>> >>>> Driver stacktrace: >>>> at org.apache.spark.scheduler.DAGScheduler.org >>>> $apache$spark$scheduler$DAGScheduler$$failJobAndIndependentStages(DAGScheduler.scala:1214) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1203) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler$$anonfun$abortStage$1.apply(DAGScheduler.scala:1202) >>>> at >>>> scala.collection.mutable.ResizableArray$class.foreach(ResizableArray.scala:59) >>>> at scala.collection.mutable.ArrayBuffer.foreach(ArrayBuffer.scala:47) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler.abortStage(DAGScheduler.scala:1202) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler$$anonfun$handleTaskSetFailed$1.apply(DAGScheduler.scala:696) >>>> at scala.Option.foreach(Option.scala:236) >>>> at >>>> org.apache.spark.scheduler.DAGScheduler.handleTaskSetFailed(DAGScheduler.scala:696) >>>> at >>>> org.apache.spark.scheduler.DAGSchedulerEventProcessActor$$anonfun$receive$2.applyOrElse(DAGScheduler.scala:1420) >>>> at akka.actor.ActorCell.receiveMessage(ActorCell.scala:498) >>>> at akka.actor.ActorCell.invoke(ActorCell.scala:456) >>>> at akka.dispatch.Mailbox.processMailbox(Mailbox.scala:237) >>>> at akka.dispatch.Mailbox.run(Mailbox.scala:219) >>>> at >>>> akka.dispatch.ForkJoinExecutorConfigurator$AkkaForkJoinTask.exec(AbstractDispatcher.scala:386) >>>> at scala.concurrent.forkjoin.ForkJoinTask.doExec(ForkJoinTask.java:260) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinPool$WorkQueue.runTask(ForkJoinPool.java:1339) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinPool.runWorker(ForkJoinPool.java:1979) >>>> at >>>> scala.concurrent.forkjoin.ForkJoinWorkerThread.run(ForkJoinWorkerThread.java:107) >>>> >>>> The same code snippet works well if the topic is with a single >>>> partition. The issue happens when the topic is using multiple partitions. >>>> Am I doing something wrong ? Can you help me find the right way to write >>>> this with kafka topic with multiple partitions. >>>> >>>> Regards, >>>> Nicolas PHUNG >>>> >>> >>> >> >
