Spark driver端得到executor返回值的方法

本文将详细解释如何从Spark驱动程序端的执行器获取返回值。这篇文章的内容质量很高，所以边肖会分享给大家参考。希望你看完这篇文章后有所了解。

有人说火花的代码不优雅，这波受不了。实际上，spark代码不优雅的主要原因是不熟悉scala。我认为spark代码仍然是值得一读的伟大大数据框架之一。

今天的文章不是为了争论spark代码是否优雅，而是主要谈一些我们在了解Spark源代码后可以使用的技巧。

使用spark时，总会有一些需求大不相同。例如，一位高尔夫球手提出了这样的需求：

白健，我想在驱动端执行任务的时候得到执行器返回的结果。例如，任务是一个规则引擎，我想知道每个规则命中多少条数据。我该怎么做？

这不是很闷热很普通吗？输出后，可以在mysql中运行sql，这是理所当然的，但这通常会更麻烦。而且有时候，数据可能会用在驱动上？具体应该怎么做？

大部分想法估计是collect方法，那么如何用collect实现呢？你可以自己想想。我只能告诉你这不容易。最好将其输出到数据库，然后驱动程序编写sql进行分析。

另一个考虑是使用自定义累加器。这样的话，结果可以在执行器端累加，然后在驱动端使用，但是具体实现起来也很麻烦。你也可以自己想办法~

然后，波浪提示会给你介绍一个比较常用和淫荡的操作。

实际上，我们在这个操作中首先想到的应该是count函数，因为返回给驱动程序然后聚合的是任务的返回值。我们可以点击想法计数功能，你可以看到

defcount(): long=sc . runjob(this，Utils.getIteratorSize _)。总和

也就是sparkcontext的runJob方法。

Utils.getIteratorSize _该方法主要统计每个迭代器中的元素个数，即每个分区中的元素个数，返回值为元素个数：

/* * * countsethnumberofelementtsfaniteratorsingawhileloopratherhancalling *[[Scala . collection . iterator # size]]因为它是一个循环，它在安全当前版本中是一个轻流*。*/defgetIteratorSize[T](iterator : iterator[T]): long={ varcount=0 while(iterator . hasnext){ count=1 iterator . next()} count }

然后runJob返回一个数组，每个数组的元素就是我们任务执行函数的返回值，然后我们调用sum得到我们的统计量。

然后我们可以用这个想法来实现我们最初的目标。这是直接在波峰上的例子：

import org . Apache . spark . { SparkConf，SparkContext，task context } import org . elastic search . Hadoop . CFG . configuration options object es2 sparkrunjob { def main(args : Array[String]): Unit={ val conf=new SparkConf()。setMaster('local[*]')。setAppName(this . getclass . getcanonicalname)conf . set(configuration options。ES_NODES，' 127 . 0 . 0 . 1 ')conf . set(configuration options。ES_PORT，' 9200 ')配置集(配置选项。“真”)配置集(配置选项。ES_INDEX_AUTO_CREATE，' true ')配置集(配置选项。ES_NODES_DISCOVERY，' false ')conf . set(' ES . write . rest . error . handlers '，' ignorrecifact ')conf . set(' ES . write . rest . error . handler . ignorrecifitshandler ')val sc=new SparkContext(conf)import org . elastic search . spark . _ val rdd=sc . esjsonrdd(' post ')。重分区(10)rdd . count()val func=(itr : Iterator[(String，String)]={ var count=0 itr . foreach(每个={ count=1 })(taskscontext . getpartitionid()，count) } val res=sc.runJob(rdd，func)RES . foreach(println)sc . stop()} }

在这个例子中，驱动程序得到的是每个任务处理的数据量。

下面是如何从Spark驱动程序端的执行器获取返回值。希望