For example, task A needs to output the calculation results every 5s, while task B USES the time window to calculate the results every 1hour. When task B is run, task A is blocked if all of it is serialized. Maybe it takes 3 minutes for B to execute, then the calculation results of A can't be seen in real time in these three minutes. Within the Spark program (that is, within each Application), tasks can run in parallel.
sparkConf.set("spark.streaming.concurrentJobs", "5")
sparkConf.set("spark.scheduler.mode", "FAIR")
I set the spark conf like this, but then something went wrong：
18/09/15 16:24:20 ERROR Executor: Exception in task 2.0 in stage 0.0 (TID 18)
java.util.ConcurrentModificationException: KafkaConsumer is not safe for multi-threaded access
at org.apache.kafka.clients.consumer.KafkaConsumer.acquire(KafkaConsumer.java:2215)
at org.apache.kafka.clients.consumer.KafkaConsumer.acquireAndEnsureOpen(KafkaConsumer.java:2199)
at org.apache.kafka.clients.consumer.KafkaConsumer.seek(KafkaConsumer.java:1505)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.seek(CachedKafkaConsumer.scala:132)
at org.apache.spark.streaming.kafka010.CachedKafkaConsumer.get(CachedKafkaConsumer.scala:67)
at org.apache.spark.streaming.kafka010.KafkaRDDIterator.next(KafkaRDD.scala:271)
at org.apache.spark.streaming.kafka010.KafkaRDDIterator.next(KafkaRDD.scala:231)
at org.apache.spark.storage.memory.MemoryStore.putIteratorAsBytes(MemoryStore.scala:379)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1109)
at org.apache.spark.storage.BlockManager$$anonfun$doPutIterator$1.apply(BlockManager.scala:1083)
at org.apache.spark.storage.BlockManager.doPut(BlockManager.scala:1018)
at org.apache.spark.storage.BlockManager.doPutIterator(BlockManager.scala:1083)
at org.apache.spark.storage.BlockManager.getOrElseUpdate(BlockManager.scala:809)
How can I do? thanks.