spark使用心得_使用spark系统的心得体会

spark入门

启停spark

sbin/start-all.sh

sbin/stop-all.sh
1
2
3

spark-shell

进入spark/bin目录，执行：
./spark-shell
输出中有这么一行：

Spark context Web UI available at http://xx.xx.xx.188:4040
1

意味着我们可以从web页面查看spark的运行情况,特别要注意的是，我们可从中看到节点的classpath，了解每个节点自带了哪些jar包。
如要查看集群各节点的信息，也可以查看http://xx.xx.xx.188:8080.

spark-shell默认运行于本地，要想运行于集群，需加上–master参数：

./spark-shell --master spark://xx.xx.xx.188:7077
1

helloworld

scala代码是：

object HelloWorld{
  def main(args: Array[String]): Unit = {
    //配置Spark应用名称
    val conf = new SparkConf().setAppName("CollectFemaleInfo")

    // 提交spark作业
    val sc = new SparkContext(conf)

    //读取数据。其是传入参数args(0)指定数据路径
    val text = sc.textFile(args(0))

    //筛选女性网民上网时间数据信息
    val data = text.filter(_.contains("female"))

    // 汇总每个女网民上网时间
    val femaleData:RDD[(String, Int)] = data.map { line =>
      val t = line.split(',')
      (t(0), t(2).toInt)
    }.reduceByKey(_+_)

    // 筛选出时间大于两小时的女网民
    val result = femaleData.filter(line => line._2 > 120)
    println("result count: " + result.count())
    result.collect().foreach(println)
  }
}
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26

注意：
1、sparkContext的textFile默认加载hdfs的文件，要处理本地文件，需加上file://前缀。用本地数据文件的话，要求spark集群里的每个节点上都有这个本地文件，否则会报文件找不到的错误。所以，方便起见，最好是将数据放到hdfs上。
2、RDD的转换（transformation，例如map、flatMap、filter等）操作都是lazy的（亦即，只是创建一个新的RDD实例，而未做任何实际计算），只有count、collect这样的行动（action）操作才会真正去求值。这跟java stream的表现是一样的。

提交jar包

将代码用maven打包为jar，接着提交给spark运行，提交本地执行的命令如下：

./spark-submit --class com.lee.ConsistencyCheck /export/home/data/com.lee.distrulechecker.service-1.0-SNAPSHOT.jar ExtArea
1

提交spark集群运行的命令如下：

./spark-submit --class com.lee.ConsistencyCheck --master spark://xx.xx.xx.188:7077  /export/home/data/com.lee.distrulechecker.service-1.0-SNAPSHOT.jar ExtArea

./spark-submit --class com.lee.ConsistencyCheck --master spark://xx.xx.xx.188:7077 --conf spark.cores.max=5 /export/home/data/com.lee.distrulechecker.service-1.0-SNAPSHOT.jar ExtArea

./spark-submit --class com.lee.ConsistencyCheck --master local /export/home/data/com.lee.distrulechecker.service-1.0-SNAPSHOT.jar ExtArea

1
2
3
4
5
6

注意:
1、–master 指定集群URL，支持的选项如下：

local 本地单线程

local[K] 本地多线程（指定K个内核）

local[*] 本地多线程（指定所有可用内核）

spark://HOST:PORT  连接到指定的 Spark standalone cluster master，需要指定端口。

mesos://HOST:PORT  连接到指定的  Mesos 集群，需要指定端口。

yarn-client客户端模式 连接到 YARN 集群。需要配置 HADOOP_CONF_DIR。

yarn-cluster集群模式 连接到 YARN 集群

1
2
3
4
5
6
7
8
9
10
11
12
13
14

如果不指定–master选项默认就在local跑。

2、内存不够可用

--driver-memory 512M --executor-memory 512M
1

强制限制内存。

修改日志打印级别

spark-submit提交时会打印很多INFO信息，影响结果查看，可通过修改日志级别解决。
spark/conf目录下复制log4j.properties.template为log4j.properties，修改：
log4j.rootCategory=INFO, console
为
log4j.rootCategory=WARN, console
则在用spark-submit提交后不会出现大量的INFO信息。

提交集群执行时报错：Initial job has not accepted any resources

网上搜了以下，几个原因：
1、主机名和ip是否配置正确，查看/etc/hosts，同时在spark-shell里键入：
sc.getConf.getAll.foreach(println)
查看conf信息
2、内存不足，SPARK_EXECUTOR_MEMORY参数默认会使用1G内存，如果不够，可以在spark-submit里指定小于1G的数值，例如：
–executor-memory 512M
3、端口号被占用，之前的程序已运行。我的情况就是这样，spark-shell使用的集群模式，会把7077端口占用掉，导致随后的spark-submit必然失败。

提交jar时的库依赖

Java 和Scala 用户可以通过spark-submit 的–jars 标记提交独立的JAR 包依赖。当只有一两个库的简单依赖，并且这些库本身不依赖于其他库时，这种方法比较合适。但是一般Java 和Scala 的工程会依赖很多库。当你向Spark 提交应用时，你必须把应用的整个依赖传递图中的所有依赖都传给集群。为此，常规的做法是使用构建工具，生成单个大JAR 包，包含应用的所有的传递依赖。这通常被称为超级（uber）JAR 或者组合（assembly） JAR。

SparkSQL

我们可以在sparkSQL里写出比较复杂的sql，比如case when：

select case when (a.NAME <> b.EXTNAME) then 1 else 0 end from OBJ1 a join OBJ2 b on a.RID=b.RID
1

spark的python接口

spark通过py4j来做到python和java的互操作。我个人的猜测，由于spark计算的效率瓶颈应该在分布式计算上，使用python的效率未必比java或scala相差很多，就好比我们产品的程序，性能瓶颈都在sql上，用啥语言组织业务更多的出于使用方便的考量。

Spark进阶

driver/executor和master/worker的概念详解

《spark快速大数据分析》里对driver/executor和master/worker的介绍：

在分布式环境下，Spark 集群采用的是主/ 从结构。在一个Spark 集群中，有一个节点负责中央协调，调度各个分布式工作节点。这个中央协调节点被称为驱动器（Driver）节点。

与之对应的工作节点被称为执行器（executor）节点。driver节点可以和大量的executor节点进行通信，它们也都作为独立的Java 进程运行。驱动器节点和所有的执行器节点一起被称为一个Spark 应用（application）。

Spark 文档中始终使用驱动器节点和执行器节点的概念来描述执行Spark应用的进程。而主节点（master）和工作节点（worker）的概念则被用来分别表述集群管理器中的中心化的部分和分布式的部分。这些概念很容易混淆，所以要格外小心。
1
2
3
4
5

上述说法比较抽象，具化后是这样：
1、一个节点就是一个JVM进程，所以driver/executor和master/worker是四种进程；
2、master/worker进程是静态的、常驻的，spark集群起来后它们就存在了，我们在主机上执行ps命令可以看到master进程；在从机上ps，可以看到worker进程
3、driver/executor进程是动态的、随application存在的，application可以简单的认为就是用spark-submit提交的jar包。我们用 spark-submit提交jar包时，就会启动driver进程，driver进程好比监工，master进程好比总包工头，监工向总包工头提要求：“该干活了”，于是master通知它管理的小包工头（worker进程）：“来来分点活给你们干”。worker进程就会去叫醒手下的工人（同一台从机上的executor进程）：“你干这个、你干那个，手脚麻利点”。所以真正干活的是executor进程，driver还干点数据汇总的活，master/worker可都是“管理者”。application运行的时候，我们可以在从机上用ps命令看，会有好几个executor进程。这些进程由application触发启动，通过线程池运行实际的任务，等application结束，它们就会自然消亡。

因此，一个application的运行会有若干管理开销，比如数据的跨节点传输、启停executor进程、启停executor进程里的线程池等，若数据量较小，这些管理开销占的比重反而较大，得不偿失。举个例子，要处理1000条记录，3台机器，每台机器上4个executor进程，结果每个进程就处理80条记录，才开始就要结束，实在太浪费了。

spark调优参数

每个executor占用的核数

spark.executor.cores

The number of cores to use on each executor. In standalone and Mesos coarse-grained modes, setting this parameter allows an application to run multiple executors on the same worker, provided that there are enough cores on that worker. Otherwise, only one executor per application will run on each worker.
1
2
3

注意，这是每个executor可以使用的core数。所以，如果一台机器上仅有8个core且spark.executor.cores=4，那么每台机器上最多能起2个executor进程。
yarn集群下，该参数默认值为1，即每个executor进程使用一个core；standalone集群下则是该节点可用的所有core，考虑到standalone集群对application的调度默认是独占的，这个默认值就不难理解了，所以我们在各个worker上仅看到一个executor进程。

application占用的最大核数

spark.cores.max

When running on a standalone deploy cluster or a Mesos cluster in "coarse-grained" sharing mode, the maximum amount of CPU cores to request for the application from across the cluster (not from each machine). If not set, the default will be spark.deploy.defaultCores on Spark's standalone cluster manager, or infinite (all available cores) on Mesos.
1
2
3

每个executor占用的内存

spark.executor.memory

Amount of memory to use per executor process (e.g. 2g, 8g).
1
2
3

默认1g。

spark的优缺点

优点：
1、扩展性好，只需增加cpu和内存，就能在增加数据量的情况下保证性能不受较大影响。
实测中，数据量10倍增长，但耗时增长远低于10倍（当然，超过10w条记录后我们启用了多核，之前都是单核运行）。

2、资源独占（或采用静态资源分配策略）的情况下，效率始终比较稳定，不像数据库要受主键、背景数据量及统计信息的影响；

缺点：
1、比较重量级，小数据量计算的额外开销反而较大。这时设置spark.cores.max为很小的值（例如1），减少并行度，反而能提升效率。尽管如此，小数据量下相比于DB依然没有优势，两张千条记录的表连接在DB上耗时不超过1s，但在spark上仍需4s，这还不包括数据提取到hdfs的时间。

2、可能由于硬件资源有限（主要是core数），应用的并发度无法做到很高，最多不能超过总的核数。从测试情况来看，10w条记录以内的应用只需1核就能保证效率，但超过10w条，就要考虑多核了，像100w条，在10核时才能保证执行时间最短。