考虑下面一种场景,在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
最后需要注意一点,在线程里面调用的方法如果包含一些全局加载的属性,最好放在线程的成员变量里面进行初始化,否则多个线程去更改全局属性,有可能会造成一些未知的问题。
有什么问题可以扫码关注微信公众号:我是攻城师(woshigcs),在后台留言咨询。 技术债不能欠,健康债更不能欠, 求道之路,与君同行。
![](http://dl2.iteye.com/upload/attachment/0104/9948/3214000f-5633-3c17-a3d7-83ebda9aebff.jpg)