Spark如何在一個SparkContext中提交多個任務

在使用spark處理數據的時候，大多數都是提交一個job執行，然後job內部會根據具體的任務，生成task任務，運行在多個進程中，比如讀取的HDFS文件的數據，spark會載入所有的數據，然後根據block個數生成task數目，多個task運行中不同的進程中，是並行的，如果在同一個進程中一個JVM裡面有多個task，那麼多個task也可以並行，這是常見的使用方式。

考慮下面一種場景，在HDFS上某個目錄下面有10個文件，我想要同時並行的去統計每個文件的數量，應該怎麼做？ 其實spark是支持在一個spark context中可以通過多線程同時提交多個任務運行，然後spark context接到這所有的任務之後，通過中央調度，在來分配執行各個task，最終任務完成程序退出。

下面就來看下如何使用多線程提交任務，可以直接使用new Thread來創建線程提交，但是不建議這麼做，推薦的做法是通過Executors線程池來非同步管理線程，尤其是在提交的任務比較多的時候用這個會更加方便。

核心代碼如下：

def main(args: Array[String]): Unit = {

val sparkConf=new SparkConf
//實例化spark context
val sc=new SparkContext(sparkConf)
sparkConf.setAppName("multi task submit ")
//保存任務返回值
val list=new util.ArrayList[Future[String]]
//並行任務讀取的path
val task_paths=new util.ArrayList[String]
task_paths.add("/tmp/data/path1/")
task_paths.add("/tmp/data/path2/")
task_paths.add("/tmp/data/path3/")

//線程數等於path的數量
val nums_threads=task_paths.size
//構建線程池
val executors=Executors.newFixedThreadPool(nums_threads)
for(i<-0 until nums_threads){ val task= executors.submit(new Callable[String] { override def call: String ={ val count=sc.textFile(task_paths.get(i)).count//獲取統計文件數量 return task_paths.get(i)+" 文件數量： "+count } }) list.add(task)//添加集合裡面 } //遍歷獲取結果 list.asScala.foreach(result=>{
log.info(result.get)
})
//停止spark
sc.stop

}

可以看到使用scala寫的代碼比較精簡，這樣就完成了一個並行task提交的spark任務，最後我們打包完畢後，上傳到linux上進行提交，命令如下：

/opt/bigdata/spark/bin/spark-submit
--class MultiTaskSubmit
--master yarn
--deploy-mode cluster
--executor-cores 3
--driver-memory 1g
--executor-memory 1g
--num-executors 10
--jars $jars task.jar

最後需要注意一點，在線程裡面調用的方法如果包含一些全局載入的屬性，最好放在線程的成員變數裡面進行初始化，否則多個線程去更改全局屬性，有可能會造成一些未知的問題。

喜歡這篇文章嗎？立刻分享出去讓更多人知道吧！