Spark如何在一个SparkContext中提交多个任务

末蓝、 2022-04-06 09:14 230阅读 0赞

在使用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

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