我是新的Spark和我有两个长期运行阶段，正在做几乎相同的事情。下面是我的伪代码。

var metaData = sqlContext.read
  .format("com.databricks.spark.csv")
  .option("header", "true") 
  .option("inferSchema", "true") 
  .load(csvFile)

val met = broadcast(metaData.dropDuplicates(Seq("col1")))

val accessLogs = sc.textFile(logFile).filter(line => regex.pattern.matcher(line).matches).map(line => LogParser.parseLogLine(line)).toDF()

val joinOutput = accessLogs.join(met,accessLogs("col1") === met("col1"),"left_outer")

val uniqueDfNames2 = Seq("col0", "col1", "col2", "col3","col4")
val sparseFilter = joinOutput
                    .filter(joinOutput.col("col1").isNotNull)
                    .filter(joinOutput.col("col2").isNotNull)
                    .flatMap(row=>ListParser.parseLogLine(row))
sparseFilter.cache()

val uniqueCount = sparseFilter
                    .filter{r=>r.col0 != null && r.col0 != "" }
                    .map{
                          case(KeyValParse(col0,col1,col2,col3,col4,col5))=>((col0,col1,col2,col3,col4,col5),1)
                        }
                    .distinct().cache()
                    .map {case ((col0,col1,col2,col3,col4),count) => ((col0,col1,col2,col3,col4),1)
                    }
                    .reduceByKey(_+_)
                    .map {case ((col0,col1,col2,col3,col4),count) => (col0,col1,col2,col3,col4,count)
                    }
                    .toDF(uniqueDfNames: _*).cache()

val totalCount = sparseFilter
                  .map{
                        case(Parse(col0,col1,col2,col3,col4,col5))=>((col0,col1,col2,col3,col4),1)
                      }
                  .reduceByKey(_+_)
                  .map{
                        case ((col0,col1,col2,col3,col4),totcount) => (col0,col1,col2,col3,col4,totcount)
                      }
                  .toDF(uniqueDfNames2: _*)
                  .join(uniqueCount,Seq("col0", "col1", "col2", "col3"),"left")
                  .select($"col0",$"col1",$"col2",$"col3",$"unicount",$"totcount")
                  .orderBy($"unicount".desc)
                  .toDF(totalDfNames: _*)

totalCount
  .select("*")
  .write
  .format("com.databricks.spark.csv")
  .option("header", "true")
  .option("delimiter", "|")
  .save(countPath)

我在这里尝试的是根据一些参数从日志中生成unique和totalcount。
一切正常，但有这两个长期运行阶段，共享几乎相同的dag。
下面是两个阶段的镜头。
请看下面两个阶段的截图。

在完成flatmap任务之前，它们都做同样的事情。为什么这些不合并到一个阶段？为什么第11阶段会再次读取文件并再次进行所有计算，这是我猜不到的？
对于一个有10个执行器（7个内核，15gbram）的20gb数据来说，几乎需要30分钟才能完成，但我觉得这可以减少到相当低的时间。
任何指导都将不胜感激。
ps:-对不起，我的图像编辑技能：）