Spark 编程起步

喜欢ヅ旅行 2022-08-19 08:58 156阅读 0赞

Spark是一个开源的分布式计算系统，提供快速的数据分析功能。 官网地址  [http://www.spark-project.org/][http_www.spark-project.org]  据说性能高出hadoop很多(个人理解主要是因为两点：内存和cache)，而且相对更加简单，灵活。非常适合需要反复迭代的计算，比如机器学习。

RDDstrack lineage information that can be used to efficiently recomputelost data

One waySpark avoids using it is through locality-aware scheduling for RAMand disk Pre-partition the links RDD so that links for URLs withthe same hash code are on the same node

（一）运行Spark

1）构建Spark  
定位至Spark主目录，如/spark-0.9.0-incubating-bin-hadoop1, 使用GitBash打开  
设置Hadoop版本：set SPARK\_HADOOP\_VERSION=1.2.1

SBT 构建

$ SPARK_HADOOP_VERSION=1.2.1 sbt/sbt clean assembly

$ SPARK_HADOOP_VERSION=2.2.0 SPARK_YARN=true sbt/sbt assembly（集成YARN）

Maven 构建

$ export MAVEN_OPTS="-Xmx2g -XX:MaxPermSize=512M -XX:ReservedCodeCacheSize=512m"

$ mvn -Dhadoop.version=1.2.1 -DskipTests -Pspark-ganglia-lgpl clean package （集成ganglia）

$ mvn -Pyarn -Dhadoop.version=2.2.0 -Dyarn.version=2.2.0 -DskipTests clean package（集成YARN）

2）  Spark 初始化

new SparkContext(master, appName, [sparkHome], [jars])

*  master：必填项，用于指定Spark或者Mesos集群地址，或者local本地模式。地址格式如下：
 *   *  local： 本地运行一个worker进程
     *  local\[N\]：本地运行N歌worker进程
     *  spark://host:port 连接 [Sparkstandalone cluster][] 集群，默认端口为7077
     *  mesos://host:port 连接 [Mesos][] 集群
     *  连接YARN集群参考： [runningon YARN][]
 *  appName：必填项，应用程序全局名称
 *  sparkHome：可选项，部署Spark应用程序时使用，用于指定集群所有结点中Spark的安装路径，必须一致
 *  jars：可选项，部署Spark应用程序时使用，指定需要部署至集群的本地代码和相关依赖包，以逗号分隔，如：ADD\_JARS=a.jar,b.jar./bin/spark-shell

3）Spark Shell工具

在4个线程中运行 $MASTER = local \[ 4 \]  ./bin/spark-shell

在4个线程，且依赖某类库运行 $MASTER = local \[ 4 \]  ADD\_JARS = code.jar./bin/spark-shell

（二）Spark体系

1）Spark 核心抽象  
resilient distributeddataset  (RDD)：将元素集合分区贯穿于整个集群内存资源，可在其上面并行执行操作，并且高度容错。Spark的RDD分为两种形态：parallelizedcollections  ，基于现有的Scala collection集合构建；Hadoopdatasets  ，基于HDFS构建，或者其他Hadoop支持的文件存储系统。

sharedvariables：Spark通过以传递变量的方式在集群所有结点的任务子集中运行函数，有些变量需要贯穿集群结点中参与运算的所有Task任务，而有些变量需要Task任务与Master主控程序的Driver程序交互。Spark的sharedvariables分为两种形态：broadcastvariables，用于在所有结点中cache缓存一个变量；accumulators，仅用于叠加，诸如计数器或累加和。

# 2）Resilient DistributedDatasets (RDDs) #

## 2.1）ParallelizedCollections ##

使用parallelize 方法构建parallelizedcollections

scala> val data = Array(1, 2, 3, 4, 5)

scala> val distData = sc.parallelize(data)

Spark会在集群的每一个分片slice运行一个task，默认情况下会为每个CPU分配2-4个slice，也可以手动指定slice个数，如下：

scala> val distData = sc.parallelize(data,10)

## 2.2）HadoopDatasets ##

Spark可以从HDFS，或者其他Hadoop支持的文件存储系统（如：本地文件系统、AmazonS3、Hypertable、HBase）创建Distributed dataset，输入格式可以是文本文件、 [SequenceFiles][]、或者其他Hadoop输入格式。如创建一个基于文本文件的Distributeddataset

scala> val textFile = sc.textFile("README.md")

textFile.count()

## textFile同样提供了一个参数控制slice个数，默认情况下Spark为每个file block（64MB by default inHDFS）分配一个slice，当然也可以手动调整，但注意保证slices 〉= blocks ##

2.3） 运行更多例子

bin/run-example org.apache.spark.examples.SparkPi local
    
    bin/run-example org.apache.spark.examples.SparkTC local[2]

## 2.3）RDDOperations ##

### 2.3.1）Transformations ###

<table style="margin-bottom:10px; font-size:14px; font-family:'Helvetica neue',Helvetica,Arial,sans-serif; border-collapse:collapse; color:rgb(51,51,51); margin-top:10px; display:table; background-color:rgb(255,255,255)"> 
 <tbody style=""> 
  <tr style=""> 
   <th style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> Transformation</th> 
   <th style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> Meaning</th> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> map(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回一个新的分布式数据集，由每个原元素经过func函数转换后组成</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> filter(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回一个新的数据集，由经过func函数后返回值为true的原元素组成</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> flatMap(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 类似于map，但是每一个输入元素，会被映射为0到多个输出元素（因此，func函数的返回值是一个Seq，而不是单一元素）</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> mapPartitions(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 类似于map，但独立地在RDD的每一个分块上运行，因此在类型为T的RDD上运行时，func的函数类型必须是Iterator[T]=&gt; Iterator[U]</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> mapPartitionsWithIndex(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 类似于mapPartitions,但func带有一个整数参数表示分块的索引值。因此在类型为T的RDD上运行时，func的函数类型必须是(Int,Iterator[T]) =&gt; Iterator[U]</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> sample(withReplacement,fraction,&nbsp;seed)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 根据给定的随机种子seed，随机抽样出数量为fraction的数据</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> union(otherDataset)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回一个新的数据集，由原数据集和参数联合而成</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> distinct([numTasks]))</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回一个包含源数据集中所有不重复元素的新数据集</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> groupByKey([numTasks])</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> <p style="margin:5px 0px">在一个由（K,V）对组成的数据集上调用，返回一个（K，Seq[V])对的数据集。注意：默认情况下，使用8个并行任务进行分组，你可以传入numTask可选参数，根据数据量设置不同数目的Task<br style=""> （groupByKey和filter结合，可以实现类似Hadoop中的Reduce功能）</p> </td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> reduceByKey(func, [numTasks])</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 在一个（K，V)对的数据集上使用，返回一个（K，V）对的数据集，key相同的值，都被使用指定的reduce函数聚合到一起。和groupbykey类似，任务的个数是可以通过第二个可选参数来配置的。</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> sortByKey([ascending], [numTasks])</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> <p style="margin:5px 0px">在类型为( K, V)的数据集上调用，返回以K为键进行排序的（K，V）对数据集。升序或者降序由boolean型的ascendingOrder参数决定<br style=""> （类似于Hadoop的Map-Reduce中间阶段的Sort，按Key进行排序）</p> </td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> join(otherDataset, [numTasks])</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 在类型为（K,V)和（K,W)类型的数据集上调用，返回一个（K,(V,W))对，每个key中的所有元素都在一起的数据集</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> cogroup(otherDataset, [numTasks])</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 在类型为（K,V)和(K,W)类型的数据集上调用，返回一个数据集，组成元素为（K, Seq[V], Seq[W])Tuples。这个操作在其它框架，称为 groupWith.</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> cartesian(otherDataset)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 笛卡尔积。但在数据集T和U上调用时，返回一个(T，U）对的数据集，所有元素交互进行笛卡尔积。</td> 
  </tr> 
 </tbody> 
</table>

### 2.3.2）Actions ###

<table style="margin-bottom:10px; font-size:14px; font-family:'Helvetica neue',Helvetica,Arial,sans-serif; border-collapse:collapse; color:rgb(51,51,51); margin-top:10px; display:table; background-color:rgb(255,255,255)"> 
 <tbody style=""> 
  <tr style=""> 
   <th style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> Action</th> 
   <th style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> Meaning</th> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> reduce(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 通过函数func聚集数据集中的所有元素。Func函数接受2个参数，返回一个值。这个函数必须是关联性的，确保可以被正确的并发执行</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> collect()</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 在Driver的程序中，以数组的形式，返回数据集的所有元素。这通常会在使用filter或者其它操作后，返回一个足够小的数据子集再使用，直接将整个RDD集Collect返回，很可能会让Driver程序OOM</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> count()</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回数据集的元素个数</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> first()</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回数据集的第一个元素（类似于take（1））</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> take(n)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> <p style="margin:5px 0px">返回一个数组，由数据集的前n个元素组成。注意，这个操作目前并非在多个节点上，并行执行，而是Driver程序所在机器，单机计算所有的元素<br style=""> (Gateway的内存压力会增大，需要谨慎使用）</p> </td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> takeSample(withReplacement,num,&nbsp;seed)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 返回一个数组，在数据集中随机采样num个元素组成，可以选择是否用随机数替换不足的部分，Seed用于指定的随机数生成器种子</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> saveAsTextFile(path)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 将数据集的元素，以textfile的形式，保存到本地文件系统，hdfs或者任何其它hadoop支持的文件系统。Spark将会调用每个元素的toString方法，并将它转换为文件中的一行文本</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> saveAsSequenceFile(path)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 将数据集的元素，以sequencefile的格式，保存到指定的目录下，本地系统，hdfs或者任何其它hadoop支持的文件系统。RDD的元素必须由key-value对组成，并都实现了Hadoop的Writable接口，或隐式可以转换为Writable（Spark包括了基本类型的转换，例如Int，Double，String等等）</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> countByKey()</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 对(K,V)类型的RDD有效，返回一个(K，Int)对的Map，表示每一个key对应的元素个数</td> 
  </tr> 
  <tr style=""> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> foreach(func)</td> 
   <td style="border-top:rgb(153,153,153) 1px solid; height:25px; border-right:rgb(153,153,153) 1px solid; min-width:25px; border-bottom:rgb(153,153,153) 1px solid; word-break:break-all; padding-bottom:5px; padding-top:5px; padding-left:10px; min-height:25px; border-left:rgb(153,153,153) 1px solid; padding-right:10px"> 在数据集的每一个元素上，运行函数func。这通常用于更新一个累加器变量，或者和外部存储系统做交互</td> 
  </tr> 
 </tbody> 
</table>

## 2.4）RDDPersistence ##

几项选择原则：

*  如果对象可以合适存储在内存中，尽量保持默认的MEMORY\_ONLY，这可以使得CPU利用率最大化，RDD的Operation（transformation/action）可以最快
 *  如果内存大小不适合存储原始对象，那么使用MEMORY\_ONLY\_SER，并且选择一个合适的序列化工具，使得空间利用率最大化，以及可以快速访问
 *  尽量不要使用硬盘存储，除非在内存计算dataset会产生高昂的代价，或者要过滤一个巨大的数据集，否则在分区中重新计算dataset的效率都优于从硬盘加载
 *  尽可能使用副本存储机制以保证快速容错，所有的容错机制都可以保证重新计算丢失的数据，但是副本可以让你继续运行而不用等待重新计算丢失的分区数据

# 3）SharedVariables #

## 3.1）广播变量Broadcast Variables ##

## BroadcastVariables可以在集群所有节点中cache缓存一个只读的变量，而不用来回传递。BroadcastVariables一旦创建，那么任何集群中的任何函数都可以使用它，并且由于它是只读的，可以保证所有节点都有相同的值。BroadcastVariables使用broadcast创建，使用value读取，如： ##

scala> val broadcastVar = sc.broadcast(Array(1, 2, 3)) scala> broadcastVar.value

现在broadcastVar可以在集群所有结点的函数中使用了，而不用再进行传递！

## 3.2）累加器Accumulators ##

Accumulators  Variables可以用于分布式并行计算中的叠加操作，比如计数器和累加求和。Spark对Accumulators  Variables支持Int和Double类型，当然通过编码可实现自定义类型。Accumulators  Variables通过accumulator创建，+=叠加，value获取值（节点的Task无法读取，只有Driver才行）。如下代码：

scala> val accum = sc.accumulator(0) accum: spark.Accumulator[Int] = 0 scala> sc.parallelize(Array(1, 2, 3, 4)).foreach(x => accum += x) ... 10/09/29 18:41:08 INFO SparkContext: Tasks finished in 0.317106 s scala> accum.value res2: Int = 10

（三）Spark运行架构

（四）Spark部署模式

[http://mp.weixin.qq.com/s?\_\_biz=MjM5NzAyNTE0Ng==&mid=200101745&idx=1&sn=ab0413f1293cdd46990a4a4274ed1999][http_mp.weixin.qq.com_s_biz_MjM5NzAyNTE0Ng_mid_200101745_idx_1_sn_ab0413f1293cdd46990a4a4274ed1999]

[http://blog.csdn.net/colorant/article/details/18549027][http_blog.csdn.net_colorant_article_details_18549027]

*  [Amazon EC2][]: our EC2 scripts let you launch a cluster in about 5minutes
 *  [Standalone Deploy Mode][Sparkstandalone cluster]: simplest way to deploy Spark on aprivate cluster
 *  [Apache Mesos][Mesos]
 *  [Hadoop YARN][runningon YARN]

Yarn-standalone  ：SparkContext和任务都运行在Yarn集群中

Yarn-client  ：SparkConext运行在本地，task运行在Yarn集群中

1) Standalone

![Spark <wbr>编程起步][Spark_wbr]

Standalone模式在极端情况下，hash算法无法均匀分配，导致某些Worker负荷过载！

配置$SPARK\_HOME/confslaves文件，添加群集所有主机hosts

*  `sbin/start-master.sh` \- Starts a masterinstance on the machine the script is executed on.
 *  `sbin/start-slaves.sh` \- Starts a slaveinstance on each machine specified inthe `conf/slaves` file.
 *  `sbin/start-all.sh` \- Starts both a masterand a number of slaves as described above.
 *  `sbin/stop-master.sh` \- Stops the master thatwas started via the `bin/start-master.sh` script.
 *  `sbin/stop-slaves.sh` \- Stops the slaveinstances that were started via `bin/start-slaves.sh`.
 *  `sbin/stop-all.sh` \- Stops both the masterand the slaves as described above.

`2）HadoopYARN`

*  构建SPARK

安装部署YARN， [http://raseshmori.wordpress.com/2012/10/14/install-hadoop-nextgen-yarn-multi-node-cluster/][http_raseshmori.wordpress.com_2012_10_14_install-hadoop-nextgen-yarn-multi-node-cluster]

`SPARK_HADOOP_VERSION=2.2.0 SPARK_YARN=true sbt/sbtassembly`

`cd/usr/local/ims/spark/spark-0.9.0-incubating-bin-hadoop2`

注意：运行YARN时，务必确保环境变量配置正确，否则无法连接ResourceManager

*  yarn-client mode（用例测试场景）

运行 SparkPi样例

SPARK\_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar\\   
SPARK\_YARN\_APP\_JAR=examples/target/scala-2.10/spark-examples-assembly-0.9.0-incubating.jar\\   
./bin/run-example org.apache.spark.examples.SparkPiyarn-client

打开spark shell（不再需要指定Spark Master）  
SPARK\_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar\\   
SPARK\_YARN\_APP\_JAR=examples/target/scala-2.10/spark-examples-assembly-0.9.0-incubating.jar\\   
MASTER=yarn-client ./bin/spark-shell

*  yarn-standalone mode（生产环境推荐）

`运行SparkPi样例`

`SPARK_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar\`

`   ./bin/spark-classorg.apache.spark.deploy.yarn.Client \`

`     --jarexamples/target/scala-2.10/spark-examples-assembly-0.9.0-incubating.jar\`

`     --classorg.apache.spark.examples.SparkPi \`

`     --args yarn-standalone\`

`     --num-workers 3\`

`     --master-memory 4g\`

`     --worker-memory 2g\`

`     --worker-cores1`

`运行SparkTC样例（观察跟踪ApplicationMasterTracking UI）`

`SPARK_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar\`

`   ./bin/spark-classorg.apache.spark.deploy.yarn.Client \`

`     --jarexamples/target/scala-2.10/spark-examples-assembly-0.9.0-incubating.jar\`

`     --classorg.apache.spark.examples.SparkTC \`

`     --args yarn-standalone\`

`     --num-workers 3\`

`     --master-memory 4g\`

`     --worker-memory 2g\`

`     --worker-cores1`

运行官网Simple APP 例子

SPARK\_JAR=./assembly/target/scala-2.10/spark-assembly-0.9.0-incubating-hadoop2.2.0.jar\\

./bin/spark-classorg.apache.spark.deploy.yarn.Client \\

\--jar /usr/local/msa\_spark/simpleApp.jar \\

\--class me.learn.spark.SimpleApp \\

\--args yarn-standalone \\

\--num-workers 3 \\

\--master-memory 4g \\

\--worker-memory 2g \\

\--worker-cores 1

（五）性能

1）度量

Ganglia： ![metrics.properties][Spark_wbr]   ![gmetad.conf][Spark_wbr]   ![gmond.conf][Spark_wbr]

2）调优

*  序列化：Kyro
 *  内存优化：

测量数据对内存容量需求，创建RDD，并cache，观察driver内存占用率

*  数据结构：

1.     使用原始对象类型，不用使用包装的wrapper对象
2.     使用对象数据，而不是集合类，万不得已可以考虑fastutil
3.     尽量避免大量的内嵌对象和指针引用
4.     使用枚举对象替代string
5.     设置JVM`-XX:+UseCompressedStrings` 对ASCIIstring使用8bit
6.     调整存储级别，优先内存，其次考虑硬盘

*     GC调优

1.     通过设置JVM的GCverbose参数，观察JVM状态，以及测量GC发生频率和耗时
2.     在worker端一个task的执行期间多次发生fullgc，那么显然内存不足以使任务高效运行
3.     调整RDD缓存容量，默认使用worker内存的66%进行old full gc，可根据old大小进行调整占比，从而增大RDD缓存利用率，避免任务执行变慢
4.     如果时常发生小规模GC，很少Full GC，那么调整young的Eden大小
5.    当有时发生OutOfMemory错误，表示task所要处理的数据集对当前节点太大了，可以尝试增加增加parallel
6.     当某些静态数据存在时，尽可能使用广播变量，以避免串行任务执行

补充：GC的过程，Java HeapSpace分作Young generation和Old generation，young存放短生命期对象，old存放长周期对象，young进而划分为Eden、Survivor，当Eden满时发生Scavengegc，对象逐步清理分别移入Eden->Survivor->Old；当Old满时进行full GC。

*     MapReduce

1.  当两个RDD做Map时，可考虑将数据集小的RDD收集collect至本地后再进行map，以防止整个集群shuffle，更进一步的方法是使用广播变量传递
2.  根据hash算法，把相同hash值数据（PageRank中的links/URLs）分配到相同节点内存进行计算，从而避免全局盲目的shuffle
3.  设置合理的partition对数据进行切片，调整合适的并发执行度
4.  基于YARN资源调度，超越集群节点一致性hash对数据在极端情况下的不均匀，而是通过ResourceManager调度

（六）提交作业（Spark standalone）

nohup sparkmsa\_wcdma\_spark.jarcom.certus.msa.netElement.NetElementFluxCountTaskspark://hadoop-1.certus.com:7077 20 >nohup.lsy.out 2>&1&

nohup sparkmsa\_wcdma\_spark.jar com.certusnet.spark.scheduler.ConfiguableJobnetelement spark://hadoop-1.certus.com:7077 >nohup.lsy.out2>&1 &

（七）常见问题

Spark Standalone集群提交任务发生：

Initial job has not accepted any resources; checkyour cluster UI to ensure that workers are registered and havesufficient memory

登陆Spark Master UI,http://dev1.msa.certusnet:8080/ 确保Worker alive，有足够内存，有足够Cores，检查$SPARK\_HOME/conf/spark-env.sh,确保SPARK\_WORKER\_MEMORY启动配置合理，并且应用程序申请合理

[http_www.spark-project.org]: http://www.spark-project.org/
[Sparkstandalone cluster]: https://spark.incubator.apache.org/docs/latest/spark-standalone.html
[Mesos]: https://spark.incubator.apache.org/docs/latest/running-on-mesos.html
[runningon YARN]: https://spark.incubator.apache.org/docs/latest/running-on-yarn.html
[SequenceFiles]: http://hadoop.apache.org/common/docs/current/api/org/apache/hadoop/mapred/SequenceFileInputFormat.html
[http_mp.weixin.qq.com_s_biz_MjM5NzAyNTE0Ng_mid_200101745_idx_1_sn_ab0413f1293cdd46990a4a4274ed1999]: http://mp.weixin.qq.com/s?__biz=MjM5NzAyNTE0Ng==&mid=200101745&idx=1&sn=ab0413f1293cdd46990a4a4274ed1999
[http_blog.csdn.net_colorant_article_details_18549027]: http://blog.csdn.net/colorant/article/details/18549027
[Amazon EC2]: https://spark.incubator.apache.org/docs/latest/ec2-scripts.html
[Spark_wbr]: 
[http_raseshmori.wordpress.com_2012_10_14_install-hadoop-nextgen-yarn-multi-node-cluster]: http://raseshmori.wordpress.com/2012/10/14/install-hadoop-nextgen-yarn-multi-node-cluster/