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

栏目: 编程工具 · 发布时间: 6年前

内容简介：有人说spark的代码不优雅，这个浪尖就忍不了了。实际上，说spark代码不优雅的主要是对scala不熟悉，spark代码我觉得还是很赞的，最值得阅读的大数据框架之一。今天这篇文章不是为了争辩Spark 代码优雅与否，主要是讲一下理解了spark源码之后我们能使用的一些小技巧吧。

有人说spark的代码不优雅，这个浪尖就忍不了了。实际上，说spark代码不优雅的主要是对scala不熟悉，spark代码我觉得还是很赞的，最值得阅读的大数据框架之一。

今天这篇文章不是为了争辩Spark 代码优雅与否，主要是讲一下理解了spark源码之后我们能使用的一些小技巧吧。

spark 使用的时候，总有些需求比较另类吧，比如有球友问过这样一个需求：

浪尖，我想要在driver端获取executor执行task返回的结果，比如task是个规则引擎，我想知道每条规则命中了几条数据，请问这个怎么做呢？

这个是不是很骚气，也很常见，按理说你输出之后，在 mysql 里跑条 sql 就行了，但是这个往往显的比较麻烦。而且有时候，在 driver可能还要用到这些数据呢？具体该怎么做呢？

大部分的想法估计是collect方法，那么用collect如何实现呢？大家自己可以考虑一下，我只能告诉你不简单，不如输出到数据库里，然后driver端写sql分析一下。

还有一种考虑就是使用自定义累加器。这样就可以在executor端将结果累加然后在driver端使用，不过具体实现也是很麻烦。大家也可以自己琢磨一下下～

那么，浪尖就给大家介绍一个比较常用也比较骚的操作吧。

其实，这种操作我们最先想到的应该是count函数，因为他就是将task的返回值返回到driver端，然后进行聚合的。我们可以从idea count函数点击进去，可以看到

  def count(): Long = sc.runJob(this, Utils.getIteratorSize _).sum

也即是sparkcontext的runJob方法。

Utils.getIteratorSize _这个方法主要是计算每个iterator的元素个数，也即是每个分区的元素个数，返回值就是元素个数：

/**

* Counts the number of elements of an iterator using a while loop rather than calling

* [[scala.collection.Iterator#size]] because it uses a for loop, which is slightly slower

* in the current version of Scala.

*/

def getIteratorSize[T](iterator: Iterator[T]): Long = {

var count = 0L

while (iterator.hasNext) {

count += 1L

iterator.next()

}

count

}

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

那么我们完全可以借助这个思路实现我们开头的目标。浪尖在这里直接上案例了：

import org.apache.spark.{SparkConf, SparkContext, TaskContext}

import org.elasticsearch.hadoop.cfg.ConfigurationOptions

object es2sparkRunJob {

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

val conf = new SparkConf().setMaster("local[*]").setAppName(this.getClass.getCanonicalName)

conf.set(ConfigurationOptions.ES_NODES, "127.0.0.1")

conf.set(ConfigurationOptions.ES_PORT, "9200")

conf.set(ConfigurationOptions.ES_NODES_WAN_ONLY, "true")

conf.set(ConfigurationOptions.ES_INDEX_AUTO_CREATE, "true")

conf.set(ConfigurationOptions.ES_NODES_DISCOVERY, "false")

conf.set("es.write.rest.error.handlers", "ignoreConflict")

conf.set("es.write.rest.error.handler.ignoreConflict", "com.jointsky.bigdata.handler.IgnoreConflictsHandler")

val sc = new SparkContext(conf)

import org.elasticsearch.spark._

val rdd = sc.esJsonRDD("posts").repartition(10)

rdd.count()

val func = (itr : Iterator[(String,String)]) => {

var count = 0

itr.foreach(each=>{

count += 1

})

(TaskContext.getPartitionId(),count)

}

val res = sc.runJob(rdd,func)

res.foreach(println)

sc.stop()

}

}

例子中driver端获取的就是每个task处理的数据量。

效率高，而且操作灵活高效～

是不是很骚气～～

更多spark源码知识，flink知识，欢迎加入浪尖知识星球，一起学习～

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

以上就是本文的全部内容，希望对大家的学习有所帮助，也希望大家多多支持码农网

查看所有标签

猜你喜欢:

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。

码农书籍

C语言的科学和艺术

罗伯茨 / 翁惠玉 / 机械工业出版社 / 2005-3 / 55.00元

《C语言的科学和艺术》是计算机科学的经典教材，介绍了计算机科学的基础知识和程序设计的专门知识。《C语言的科学和艺术》以介绍ANSI C为主线，不仅涵盖C语言的基本知识，而且介绍了软件工程技术以及如何应用良好的程序设计风格进行开发等内容。《C语言的科学和艺术》采用了库函数的方法，强调抽象的原则，详细阐述了库和模块化开发。此外，《C语言的科学和艺术》还利用大量实例讲述解决问题的全过程，对开发过程中常见......一起来看看《C语言的科学和艺术》这本书的介绍吧!

码农工具

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

C语言的科学和艺术

JSON 在线解析

Markdown 在线编辑器

html转js在线工具