title: spark streaming can't start multiple threads
body: <p>I use spark streaming to receive Kafka's data like this:</p>
  
  <pre><code>val conf = new SparkConf()
  conf.setMaster("local[*]").setAppName("KafkaStreamExample")
    .setSparkHome("/home/kufu/spark/spark-1.5.2-bin-hadoop2.6")
    .setExecutorEnv("spark.executor.extraClassPath","target/scala-2.11/sparkstreamexamples_2.11-1.0.jar")
  
  val threadNum = 3
  
  val ssc = new StreamingContext(conf, Seconds(2))
  val topicMap = Map(consumeTopic -&gt; 1)
  
  val dataRDDs:IndexedSeq[InputDStream[(String, String)]] = approachType match {
    case KafkaStreamJob.ReceiverBasedApproach =&gt;
      (1 to threadNum).map(_=&gt;
        KafkaUtils.createStream(ssc, zkOrBrokers, "testKafkaGroupId", topicMap))
    case KafkaStreamJob.DirectApproach =&gt;
      (1 to threadNum).map(_=&gt;
        KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder](
          ssc, Map[String, String]("metadata.broker.list" -&gt; zkOrBrokers),
        Set[String](consumeTopic)))
  }
  
  //dataRDDs.foreach(_.foreachRDD(genProcessing(approachType)))
  val dataRDD = ssc.union(dataRDDs)
  dataRDD.foreachRDD(genProcessing(approachType))
  
  ssc.start()
  ssc.awaitTermination()
  </code></pre>
  
  <p>the genProcessing generates a process to deal with the data, which will takes 5s(sleep 5s). Code are like this:</p>
  
  <pre><code>def eachRDDProcessing(rdd:RDD[(String, String)]):Unit = {
    if(count&gt;max) throw new Exception("Stop here")
    println("--------- num: "+count+" ---------")
  
    val batchNum = count
    val curTime = System.currentTimeMillis()
  
    Thread.sleep(5000)
  
    val family = approachType match{
      case KafkaStreamJob.DirectApproach =&gt; KafkaStreamJob.DirectFamily
      case KafkaStreamJob.ReceiverBasedApproach =&gt; KafkaStreamJob.NormalFamily
    }
  
    val families = KafkaStreamJob.DirectFamily :: KafkaStreamJob.NormalFamily :: Nil
  
    val time = System.currentTimeMillis().toString
  
    val messageCount = rdd.count()
  
    rdd.foreach(tuple =&gt; {
      val hBaseConn = new HBaseConnection(KafkaStreamJob.rawDataTable,
        KafkaStreamJob.zookeeper, families)
      hBaseConn.openOrCreateTable()
      val puts = new java.util.ArrayList[Put]()
      val strs = tuple._2.split(":")
      val row = strs(1) + ":" + strs(0) + ":" + time
      val put = new Put(Bytes.toBytes(row))
      put.add(Bytes.toBytes(family), Bytes.toBytes(KafkaStreamJob.tableQualifier),
        Bytes.toBytes("batch " + batchNum.toString + ":" + strs(1)))
      puts.add(put)
      hBaseConn.puts(puts)
      hBaseConn.close()
    })
  
    count+=1
    println("--------- add "+messageCount+" messages ---------")
  }
  eachRDDProcessing
  </code></pre>
  
  <p>but the spark streaming doesn't start multi-thread.Tasks were processed one by one, and each task took around 5s. My machine has 8 cores, and the spark run on one node.</p>