通过可视化来了解你的Spark应用程序
发表于 16:49|
作者Andrew Or
摘要：在过去，Spark UI一直是用户应用程序调试的帮手。而在最新的Spark 1.4版本中，一个新的因素被注入到Spark UI——数据可视化。
【编者按】在""一文中，我们有简单地介绍了1.4版本给Spark注入的新特性，在各个组件的介绍中也提到了新UI给用户带来的便捷。而从本文开始，我们将通过Databricks Blog上的系列文章深入了解新版本中的数据可视化，首先分享的是这个系列的第一篇博文——，作者&Andrew Or。以下为译文图片最大的价值就是它可以让我们发现未曾预期的事情——John Tukey。
在过去，Spark UI一直是用户应用程序调试的帮手。而在中，我们很高兴地宣布，一个新的因素被注入到Spark
UI——数据可视化。在此版本中，可视化带来的提升主要包括三个部分：
Spark events时间轴视图
&Execution DAG
Spark Streaming统计数字可视化
我们会通过一个系列的两篇博文来介绍上述特性，本次则主要分享前两个部分——Spark events时间轴视图和Execution DAG。Spark
Streaming统计数字可视化将在下一篇博文中解释。
events时间轴视图
从Spark 初期版本至今，Spark events一直是面向用户API的一部分。在最新的1.4版本，Spark UI将会把这些events在一个时间轴中显示，让用户可以一眼区别相对和交叉顺序。
时间轴视图可以覆盖3个等级：所有Job，指定的某个Job，以及指定的某个stage。在下图中，时间轴显示了横跨一个应用程序所有作业中的Spark
这里的events顺序相对简单，在所有 executors 注册后，在应用程序并行运行的4个job中，有一个失败，其余成功。当所有工作完成，并在应用程序退出后，executors同样被移除。下面不妨点击关注其中的一个job：
该job在3个文件中做word count，最后join并输出结果。从时间轴上看，很明显， 3个 word count stages 并行运行，因为它们不互相依赖。同时，最后一个阶段需要依赖前3个文件word
count的结果，所以相应阶段一直等到所有先行阶段完成后才开始。下面着眼单个stage：
这个stage被切分为20个partitions，分别在4台主机上完成（图片并没有完全显示）。每段代表了这个阶段的一个单一任务。从这个时间轴来看，我们可以得到这个stage上的几点信息。
首先，partitions在机器中的分布状态比较乐观。其次，大部分的任务执行时间分配在原始的计算上，而不是网络或I/ O开销。这并不奇怪，因为传输的数据很少。最后，我们可以通过给executors分配更多的核心来提升并行度；从目前来看，每个executors可以同时执行不超过两个任务。
借此机会展示一下Spark通过该时间轴获得的另一个特性——动态分配。该特性允许Spark基于工作负载来动态地衡量executors 的数量，从而让集群资源更有效地共享。不妨看向下张图表：
首先要注意的是，这个应用程序是在工作的过程中获得executors ，而不是预先分配好。在第一个job结束后，用于该job的executors将闲置并返回到集群。因此在这个期间，同集群中运行的其他应用程序可以获得这些资源，从而增加集群资源利用率。只有当一个新的job执行时，Spark应用程序才会获取一组新的executors
来运行它。
在一个时间轴中查看Spark events的能力有助于确定应用程序瓶颈，从而在调试过程中进行更有针对性的优化。
Execution DAG
在新版本的Spark中，第二个可视化聚焦DAG执行的每个作业。在Spark中，job与被组织在DAG中的一组RDD依赖性密切相关，类似下图：
这个job执行一个简单的word cout。首先，它执行一个textFile从HDFS中读取输入文件，然后进行一个flatMap操作把每一行分割成word，接下来进行一个map操作，以形成form（word，1）对，最后进行一个reduceByKey操作总结每个word的数值。
可视化的蓝色阴影框对应到Spark操作，即用户调用的代码。每个框中的点代表对应操作下创建的RDDs。操作本身由每个流入的stages划分。
通过可视化我们可以发现很多有价值的地方。首先，根据显示我们可以看出Spark对流水线操作的优化——它们不会被分割。尤其是，从HDF
S读取输入分区后，每个executor随后即对相同任务上的partion做flatMap和map，从而避免与下一个stage产生关联。
其次，RDDs在第一个stage中会进行缓存（用绿色突出表示），从而避免对HDFS（磁盘）相关读取工作。在这里，通过缓存和最小化文件读取可以获得更高的性能。
DAG可视化的价值在复杂jobs中体现的尤为明显。比如下图中的ALS计算，它会涉及到大量的map、join、groupByKey操作。
值得注意的是，在ALS中，缓存准确性将对性能产生的影响非常大，因为该算法在每次迭代中会重度使用之前步骤产生的结果。如今通过DAG可视化，用户和开发人员可以一目了然地查明RDDS是否被恰当地缓存，如果没有，可以快速理理解实现缓慢的原因。
与时间轴视图一样，DAG可视化允许用户点击进入一个stage进行更详细地观察。下图描述了ALS中一个独立的stage。
在stage视图中，属于这个stage的所有RDDS细节被自动展开。当前，用户可以快速地找到具体的RDDS信息，而不必job页面通过悬停各个点来猜测和检查。
最后，在这里突出一下DAG可视化和
SparkSQL之间的一个初步的集成。对比更接近物理实体层面的Spark操作，Spark SQL用户显然更熟悉一些高级操作，因此一些高级操作更需要被可视化。其结果类似将一个SQL查询计划映射到底层执行的DAG。
与SparkStreaming的整合在Spark 1.4版本中同样有所实现，这里在下一篇博文中会详细介绍。
在不久的将来，Spark UI可以更理解一些更高级别的函数库语义，以提供更多相关细节。
同时，Spark SQL将与Spark Streaming一样获得类似的标签。而在Spark Core中，当用户查看RDD时，类似partitions数量、调用点、缓存率都将会被可视化。
在此感谢社区中所有对可视化工作有所贡献的组织和个人，更特别感谢NTT Data的@sarutak在时间轴可视化特性中的主要贡献。英文原文：（翻译/王辉 &责编/仲浩）&【预告】&将于7月26-27日在北京友谊宾馆召开。机器学习与模式识别、大数据的机遇与挑战、人工智能与认知科学、智能机器人四个主题专家云集。人工智能产品库将同步上线，预约咨询：QQ：。欢迎关注。&
推荐阅读相关主题：
CSDN官方微信
扫描二维码,向CSDN吐槽
微信号：CSDNnews
相关热门文章很奇葩的情况,start-all.sh一切正常,但是在浏览器中,没有worker_spark吧_百度贴吧
&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&&签到排名：今日本吧第个签到，本吧因你更精彩，明天继续来努力！
本吧签到人数：0成为超级会员，使用一键签到本月漏签0次！成为超级会员，赠送8张补签卡连续签到：天&&累计签到：天超级会员单次开通12个月以上，赠送连续签到卡3张
关注：1,756贴子：
很奇葩的情况,start-all.sh一切正常,但是在浏览器中,没有worker收藏
vanxd@master:~/i/Spark/spark-1.3.1-bin-hadoop2.6/sbin& ./start-all.sh starting org.apache.spark.deploy.master.Master, logging to /home/vanxd/i/Spark/spark-1.3.1-bin-hadoop2.6/sbin/../logs/spark-vanxd-org.apache.spark.deploy.master.Master-1-master.outSlave-1: starting org.apache.spark.deploy.worker.Worker, logging to /home/vanxd/i/Spark/spark-1.3.1-bin-hadoop2.6/sbin/../logs/spark-root-org.apache.spark.deploy.worker.Worker-1-Slave-1.out这个是启动成功了吧?但是为什么没有呢..?我打开了worker的log:Connecting to master akka.tcp://sparkMaster@master:7077/user/Master...Retrying connection to master(attempt # 3)Connecting to master akka.tcp://sparkMaster@master:7077/user/Master...Retrying connection to master(attempt # 3)Connecting to master akka.tcp://sparkMaster@master:7077/user/Master...Retrying connection to master(attempt # 3)Connecting to master akka.tcp://sparkMaster@master:7077/user/Master...Retrying connection to master(attempt # 3)好像是连接不上master,但是我直接ssh master是可以无密码连接的啊.
“健康随e保”可保障儿童41种重大疾病。150元起/年，给孩子45万高额保障。
还需要一个sh start-all.sh
顶，和楼主情况一样，start-all.sh正常，但是web的workers为0。楼主解决了吗？
找到了，是防火墙的问题。关掉它就行了
是防火墙的问题，关闭后启动集群就正常了
登录百度帐号推荐应用
为兴趣而生，贴吧更懂你。或在windows上运行spark程序报错解决方法-hujunnan2
出现下列错误：
Exception in thread &main& java.lang.NoSuchMethodError: scala.collection.immutable.HashSet$.empty()Lscala/collection/immutable/HashS
&& &at akka.actor.ActorCell$.&init&(ActorCell.scala:336)
&& &at akka.actor.ActorCell$.&clinit&(ActorCell.scala)
&& &at akka.actor.RootActorPath.$div(ActorPath.scala:159)
&& &at akka.actor.LocalActorRefProvider.&init&(ActorRefProvider.scala:464)
&& &at akka.remote.RemoteActorRefProvider.&init&(RemoteActorRefProvider.scala:124)
&& &at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
&& &at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:57)
&& &at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
&& &at java.lang.reflect.Constructor.newInstance(Constructor.java:526)
&& &at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$2.apply(DynamicAccess.scala:78)
&& &at scala.util.Try$.apply(Try.scala:191)
&& &at akka.actor.ReflectiveDynamicAccess.createInstanceFor(DynamicAccess.scala:73)
&& &at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$3.apply(DynamicAccess.scala:84)
&& &at akka.actor.ReflectiveDynamicAccess$$anonfun$createInstanceFor$3.apply(DynamicAccess.scala:84)
&& &at scala.util.Success.flatMap(Try.scala:230)
&& &at akka.actor.ReflectiveDynamicAccess.createInstanceFor(DynamicAccess.scala:84)
&& &at akka.actor.ActorSystemImpl.liftedTree1$1(ActorSystem.scala:584)
&& &at akka.actor.ActorSystemImpl.&init&(ActorSystem.scala:577)
&& &at akka.actor.ActorSystem$.apply(ActorSystem.scala:141)
&& &at akka.actor.ActorSystem$.apply(ActorSystem.scala:118)
&& &at org.apache.spark.util.AkkaUtils$.org$apache$spark$util$AkkaUtils$$doCreateActorSystem(AkkaUtils.scala:122)
&& &at org.apache.spark.util.AkkaUtils$$anonfun$1.apply(AkkaUtils.scala:55)
&& &at org.apache.spark.util.AkkaUtils$$anonfun$1.apply(AkkaUtils.scala:54)
&& &at org.apache.spark.util.Utils$$anonfun$startServiceOnPort$1.apply$mcVI$sp(Utils.scala:1890)
&& &at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:160)
&& &at org.apache.spark.util.Utils$.startServiceOnPort(Utils.scala:1881)
&& &at org.apache.spark.util.AkkaUtils$.createActorSystem(AkkaUtils.scala:57)
&& &at org.apache.spark.SparkEnv$.create(SparkEnv.scala:241)
&& &at org.apache.spark.SparkEnv$.createDriverEnv(SparkEnv.scala:181)
&& &at org.apache.spark.SparkContext.createSparkEnv(SparkContext.scala:267)
&& &at org.apache.spark.SparkContext.&init&(SparkContext.scala:270)
&& &at cma.cimiss.SparkSQLTest$.main(SparkSQLTest.scala:14)
&& &at cma.cimiss.SparkSQLTest.main(SparkSQLTest.scala)
&& &at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
&& &at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:57)
&& &at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
&& &at java.lang.reflect.Method.invoke(Method.java:606)
&& &at com.intellij.rt.execution.application.AppMain.main(AppMain.java:140)
解决方法：
在Globe Library中的scala2.11删除，重新添加2.10.4即可