6 Spark开发指南

6.1 概述
6.2 开发环境准备
6.3 开发指引
6.4 Java代码样例
6.5 Scala代码样例
6.6 对外接口

6.1 概述

目标读者

本文档专供需要进行Spark应用开发的用户使用。本指南主要适用于具备Java和Scala开发经验的开发人员。

简介

Spark是分布式批处理系统和分析挖掘引擎，提供迭代式内存计算框架，支持多种语言(Scala/Java/Python)的应用开发能力。适用场景：

数据处理（Data Processing）：可以用来快速处理数据，兼具容错性和可扩展性。
迭代计算（Iterative Computation）：支持迭代计算，有效应对多步的数据处理逻辑。
数据挖掘（Data Mining）：在海量数据基础上进行复杂的挖掘分析，可支持各种数据挖掘和机器学习算法。

基本概念

以下为重要概念，可以帮助您减少在学习Spark框架所花费的时间，有助于Spark开发完全聚焦于实际业务。

RDD
即弹性分布数据集（Resilient Distributed Dataset），是Spark的核心概念。指的是一个只读的，可分区的分布式数据集，这个数据集的全部或部分可以缓存在内存中，在多次计算间重用。

RDD的生成：
- 从Hadoop文件系统（或与Hadoop兼容的其它存储系统）输入创建（例如HDFS）。
- 从父RDD转换得到新RDD。
RDD的存储：
- 用户可以选择不同的存储级别存储RDD以便重用（RDD有11种存储级别）。
- 当前RDD默认是存储于内存，但当内存不足时，RDD会溢出到磁盘中。
Dependency（RDD的依赖）
RDD的数据结构里很重要的一个域是对父RDD的依赖。两类依赖分别为：窄依赖和宽依赖。

图6-1 RDD的依赖
- 窄依赖：指父RDD的每一个分区最多被一个子RDD的分区所用，表现为一个父RDD的分区对应于一个子RDD的分区，和两个父RDD的分区对应于一个子RDD的分区。上图中，map/filter和union属于第一类，对输入进行协同划分（co-partitioned）的join属于第二类。
- 宽依赖：指子RDD的分区依赖于父RDD的所有分区，这是因为shuffle类操作，如上图中的groupByKey和未经协同划分的join。
窄依赖对优化很有利。逻辑上，每个RDD的算子都是一个fork/join（此join非上文的join算子，而是指同步多个并行任务的barrier）：把计算fork到每个分区，算完后join，然后fork/join下一个RDD的算子。如果直接翻译到物理实现，是很不经济的：一是每一个RDD（即使是中间结果）都需要物化到内存或存储中，费时费空间；二是join作为全局的barrier，是很昂贵的，会被最慢的那个节点拖死。如果子RDD的分区到父RDD的分区是窄依赖，就可以实施经典的fusion优化，把两个fork/join合为一个；如果连续的变换算子序列都是窄依赖，就可以把很多个 fork/join并为一个，不但减少了大量的全局barrier，而且无需物化很多中间结果RDD，这将极大地提升性能。Spark把这个叫做流水线（pipeline）优化。
Transformation和Action（RDD的操作）
对RDD的操作包含Transformation（返回值还是一个RDD）和Action（返回值不是一个RDD）两种。RDD的操作流程如图6-2所示。其中Transformation操作是Lazy的，也就是说从一个RDD转换生成另一个RDD的操作不是马上执行，Spark在遇到Transformations操作时只会记录需要这样的操作，并不会去执行，需要等到有Actions操作的时候才会真正启动计算过程进行计算。Actions操作会返回结果或把RDD数据写到存储系统中。Actions是触发Spark启动计算的动因。

图6-2 RDD操作示例
然后，来看一个简单的例子，如图6-3所示。RDD看起来与Scala集合类型没有太大差别，但它们的数据和运行模型大相迥异。
1. textFile算子从HDFS读取日志文件，返回file（作为RDD）。
2. filter算子筛出带“ERROR”的行，赋给errors（新RDD）。filter算子为一个Transformation操作。
3. cache算子把它缓存下来以备未来使用。
4. count算子返回errors的行数。count算子为一个Action操作。
图6-3 Scala样例
Transformation操作可以分为如下几种类型：
- 视RDD的元素为简单元素。
  输入输出一对一，且结果RDD的分区结构不变，主要是map。
  
  输入输出一对多，且结果RDD的分区结构不变，如flatMap（map后由一个元素变为一个包含多个元素的序列，然后展平为一个个的元素）。
  
  输入输出一对一，但结果RDD的分区结构发生了变化，如union（两个RDD合为一个，分区数变为两个RDD分区数之和）、coalesce（分区减少）。
  
  从输入中选择部分元素的算子，如filter、distinct（去除重复元素）、subtract（本RDD有、它RDD无的元素留下来）和sample（采样）。
- 视RDD的元素为Key-Value对。
  对单个RDD做一对一运算，如mapValues（保持源RDD的分区方式，这与map不同）；
  
  对单个RDD重排，如sort、partitionBy（实现一致性的分区划分，这个对数据本地性优化很重要）；
  
  对单个RDD基于key进行重组和reduce，如groupByKey、reduceByKey；
  
  对两个RDD基于key进行join和重组，如join、cogroup。
  说明：
  后三种操作都涉及重排，称为shuffle类操作。
Action操作可以分为如下几种：
- 生成标量，如count（返回RDD中元素的个数）、reduce、fold/aggregate（返回几个标量）、take（返回前几个元素）。
- 生成Scala集合类型，如collect（把RDD中的所有元素倒入 Scala集合类型）、lookup（查找对应key的所有值）。
- 写入存储，如与前文textFile对应的saveAsTextFile。
- 还有一个检查点算子checkpoint。当Lineage特别长时（这在图计算中时常发生），出错时重新执行整个序列要很长时间，可以主动调用 checkpoint把当前数据写入稳定存储，作为检查点。
Shuffle
Shuffle是MapReduce框架中的一个特定的phase，介于Map phase和Reduce phase之间，当Map的输出结果要被Reduce使用时，输出结果需要按key哈希，并且分发到每一个Reducer上去，这个过程就是shuffle。由于shuffle涉及到了磁盘的读写和网络的传输，因此shuffle性能的高低直接影响到了整个程序的运行效率。

下图清晰地描述了MapReduce算法的整个流程，其中shuffle phase是介于Map phase和Reduce phase之间。

图6-4 算法流程

概念上shuffle就是一个沟通数据连接的桥梁，实际上shuffle这一部分是如何实现的呢，下面我们就以Spark为例讲一下shuffle在Spark中的实现。

Shuffle操作将一个Spark的Job分成多个Stage，前面的stages会包括一个或多个ShuffleMapTasks，最后一个stage会包括一个或多个ResultTask。

Spark Application的结构

Spark Application的结构可分为两部分：初始化SparkCOntext和主体程序。

初始化SparkContext：构建Spark Application的运行环境。

构建SparkContext对象，如：

new SparkContext(master, appName, [SparkHome], [jars])参数介绍：
master：连接字符串，连接方式有local, yarn-cluster, yarn-client等
appName：构建的Application名称
SparkHome：集群中安装Spark的目录
jars：应用程序代码和依赖包

主体程序：处理数据

Spark shell命令

Spark基本shell命令，支持提交Spark应用。命令为：

./bin/spark-submit \--class <main-class>--master <master-url> \... # other options<application-jar> \[application-arguments]参数解释：
--class：Spark应用的类名
--master：Spark用于所连接的master，如yarn-client，yarn-cluster等
application-jar：Spark应用的jar包的路径
application-arguments：提交Spark应用的所需要的参数（可以为空）。

Spark Web UI界面
用于监控正在运行的或者历史的Spark作业在Spark框架各个阶段的细节以及提供日志显示，帮助用户更细粒度地去开发、配置和调优作业。

6.2 开发环境准备

6.2.1 Java开发环境准备

操作场景

本开发指南提供了Spark组件的样例代码和常用接口，便于开发者快速熟悉Spark并行计算框架。为了运行Spark组件的样例代码，你需要完成下面的操作。

开发环境可以搭建在Windows环境下，而运行环境（即客户端）只能部署在Linux环境下。

操作步骤

确认Spark组件已经安装，并正常运行。
对于scala开发环境，客户端机器推荐使用IDEA工具，安装要求如下：
- JDK使用1.7版本
- IntelliJ IDEA（版本：13.1.4），下载路径：http://www.jetbrains.com/idea/download/index.html。
- Scala（版本：2.10.4），下载路径：http://www.scala-lang.org/files/archive/scala-2.10.4.msi。
安装IntelliJ IDEA和Scala工具，并进行相应的配置。
1. 安装IntelliJ IDEA和Scala工具。
2. 安装Scala插件。
  1. 打开IntelliJ IDEA，如图1所示，选择“Configure”。
    图6-5 configure
  2. 如图6-6所示界面中，选择“Plugins”。
    图6-6 Plugins
  3. 如图 3 所示界面中，选择“Install plugin from disk”。
    图6-7 Install plugin from disk
  4. 在出现的搜索框中搜索Scala，并安装。
客户端机器的时间与Spark集群的时间要保持一致，时间差要小于5分钟。

FusionInsight HD集群的时间可通过登录主管理节点（集群管理IP地址所在节点）运行date命令查询。
下载Spark客户端程序到客户端机器中。
1. 登录FusionInsight Manager系统。
  在浏览器地址栏中输入访问地址，地址格式为“http://FusionInsight Manager系统的WebService浮动IP地址:8080/web”。
  
  例如，在IE浏览器地址栏中，输入“http://10.10.10.172:8080/web”。
2. 单击“Services > Spark> Download Client”，下载客户端程序到本地机器。说明：
  下载所有组件的客户端程序安装包，可通过在界面选择“Services > Download Client”进行下载。
解压缩客户端文件包。

说明：
客户端程序文件包为tar格式，可采用如下方式解压。
- Windows: 下载7zip工具，双击压缩包进行解压。
- Linux: 执行tar xf [客户端文件包]命令解压。
客户端安装。

说明：
客户端支持Linux安装。
1. 创建客户端安装目录
  例如：mkdir /opt/spark_client
2. 安装客户端
  执行6解压后的客户端安装包中的install.sh安装程序，执行该程序需要输入客户端安装的目标路径。
  
  例如：./install.sh /opt/spark_client
3. 加载环境变量。
  进入客户端的安装路径，运行source bigdata_env。
  
  例如：source /opt/spark_client/bigdata_env
4. 如果启用安全模式，需要使用业务账号进行安全认证。您可以联系管理员获取账号和密码。执行如下命令，输入密码后获取认证。
  其中spark表示业务账号的名称。
  
  kinit spark
5. 安全模式下如果要在程序中通过机机帐号进行认证，请参见《MapReduce开发指南》中的运行应用的“使用机机帐号在Linux客户端运行应用程序”部分。说明：在二次开发过程中，PRINCIPAL需要用到的用户名，应该填写为带域名的用户名，例如创建的用户为user，其PRINCIPAL用户名则为user@HADOOP.COM，代码举例：
```
conf.set(PRINCIPAL, "user@HADOOP.COM");
```
配置客户端网络连接。

说明：
当客户端所在主机不是集群中的节点时，配置客户端网络连接，可避免执行客户端命令时出现错误。
1. 确认客户端与服务端各个主机网络上互通。
2. 如果采用yarn-client模式运行Spark任务，请在客户端的“#{client_install_home}/Spark/spark/conf/spark-defaults.conf”文件中添加参数“spark.driver.host”，并将参数值设置为客户端的IP。
  如果不采用yarn-client模式运行Spark任务，则执行8.c。
3. 确认客户端中是否存在DNS服务器。
  1. 是，执行8.d。
  2. 否，执行8.e。
4. 将“hosts”文件内容导入DNS服务器，执行9。
5. 拷贝解压目录下的“hosts”文件中的内容，到客户端的hosts文件中。本地hosts文件存放路径举例：“/etc/hosts” 说明：
  - 当采用yarn-client模式时，为了Spark Web UI能够正常显示，需要在yarn的主备节点，即集群中的ResourceManager节点，将客户端的IP及主机名对应关系配置到hosts文件中。
  - 当采用yarn-client模式时，客户端的hosts文件必须配置客户端自己的IP和hostname对应关系。
导入Java代码样例工程到IDEA开发环境。
1. 在windows上的客户端安装包中存在Spark的样例工程“Spark\SparkJavaExample”。说明：
  由于样例工程在windows中搭建，所以需要把该工程拷贝到windows系统中，以备后续导入样例工程所有操作均在windows上进行。
2. 解压客户端安装包。
  运行“$ClientConfig_Path\Spark\install.bat”导入工程依赖包。
3. 单击“File > Open”，选择样例工程的SparkJavaExample文件夹，单击“OK”，工程成功导入。
4. 设置IDEA的文本文件编码格式，解决乱码显示问题。
  在IDEA的菜单栏中，选择“”图标。弹出“Settings”窗口。
  
  在左边导航上选择“File Encodings”，在“IDE encoding”区域，选择参数值为“UTF-8”，单击“Apply”后，单击“OK”，如图6-8所示。
  
  图6-8 File Encodings
（可选）使用IDEA创建一个Spark样例工程。

6.2.2 Scala开发环境准备

操作场景

本开发指南提供了Spark组件的样例代码和常用接口，便于开发者快速熟悉Spark并行计算框架。为了运行Spark组件的样例代码，你需要完成下面的操作。

开发环境可以搭建在Windows环境下，而运行环境（即客户端）只能部署在Linux环境下。

操作步骤

确认Spark组件已经安装，并正常运行。
对于scala开发环境，客户端机器推荐使用IDEA工具，安装要求如：
- JDK使用1.7版本
- IntelliJ IDEA（版本：13.1.4），下载路径：http://www.jetbrains.com/idea/download/index.html。
- Scala（版本：2.10.4），下载路径：http://www.scala-lang.org/files/archive/scala-2.10.4.msi。
安装IntelliJ IDEA和Scala工具，并进行相应的配置。
1. 安装IntelliJ IDEA和Scala工具。
2. 安装Scala插件。
  1. 打开IntelliJ IDEA，如图1所示，选择“Configure”。
    图6-12 configure
  2. 如图6-13所示界面中，选择“Plugins”。
    图6-13 Plugins
  3. 如图 3 所示界面中，选择“Install plugin from disk”。
    图6-14 Install plugin from disk
  4. 在出现的搜索框中搜索Scala，并安装。
客户端机器的时间与Spark集群的时间要保持一致，时间差要小于5分钟。

FusionInsight HD集群的时间可通过登录主管理节点（Hadoop管理IP地址所在节点）运行date命令查询。
下载Spark客户端程序到客户端机器中。
1. 登录FusionInsight Manager系统。
  在浏览器地址栏中输入访问地址，地址格式为“http://FusionInsight Manager系统的WebService浮动IP地址:8080/web”。
  
  例如，在IE浏览器地址栏中，输入“http://10.10.10.172:8080/web”。
2. 单击“Services > Spark> Download Client”，下载客户端程序到本地机器。说明：
  下载所有组件的客户端程序安装包，可通过在界面选择“Services > Download Client”进行下载。
解压缩客户端文件包。

说明：
客户端程序文件包为tar格式，可采用如下方式解压。
- Windows: 下载7zip工具，双击压缩包进行解压。
- Linux: 执行tar xf [客户端文件包]命令解压。
客户端安装。

说明：
客户端支持Linux安装。
1. 创建客户端安装目录
  例如：mkdir /opt/spark_client
2. 安装客户端
  执行6解压后的客户端安装包中的install.sh安装程序，执行该程序需要输入客户端安装的目标路径。
  
  例如：./install.sh /opt/spark_client
3. 加载环境变量。
  进入客户端的安装路径，运行source bigdata_env。
  
  例如：source /opt/spark_client/bigdata_env
4. 如果启用安全模式，需要使用业务账号进行安全认证。您可以联系管理员获取账号和密码。执行如下命令，输入密码后获取认证。
  其中spark表示业务账号的名称。
  
  kinit spark
5. 安全模式下如果要在程序中通过机机帐号进行认证，请参见《<MapReduce开发指南》的运行应用的“使用机机帐号在Linux客户端运行应用程序”部分。说明：在二次开发过程中，PRINCIPAL需要用到的用户名，应该填写为带域名的用户名，例如创建的用户为user，其PRINCIPAL用户名则为user@HADOOP.COM，代码举例：
```
conf.set(PRINCIPAL, "user@HADOOP.COM");
```
配置客户端网络连接。

说明：
当客户端所在主机不是集群中的节点时，配置客户端网络连接，可避免执行客户端命令时出现错误。
1. 确认客户端与服务端各个主机网络上互通。
2. 如果采用yarn-client模式运行Spark任务，请在客户端的“#{client_install_home}/Spark/spark/conf/spark-defaults.conf”文件中添加参数“spark.driver.host”，并将参数值设置为客户端的IP。
  如果不采用yarn-client模式运行Spark任务，则执行8.c。
3. 确认客户端中是否存在DNS服务器。
  1. 是，执行8.d。
  2. 否，执行8.e。
4. 将“hosts”文件内容导入DNS服务器，执行9。
5. 拷贝解压目录下的“hosts”文件中的内容，到客户端的hosts文件中。本地hosts文件存放路径举例：“/etc/hosts” 说明：
  - 当采用yarn-client模式时，为了Spark Web UI能够正常显示，需要在yarn的主备节点，即集群中的ResourceManager节点，将客户端的IP及主机名对应关系配置到hosts文件中。
  - 当采用yarn-client模式时，客户端的hosts文件必须配置客户端自己的IP和hostname对应关系。
导入Scala代码样例工程到IDEA开发环境。
1. 在windows上的客户端安装包中存在Spark的样例工程“Spark\SparkScalaExample”。说明：
  由于样例工程在windows中搭建，所以需要把该工程拷贝到windows系统中，以备后续导入样例工程所有操作均在windows上进行。
2. 解压客户端安装包。
  运行“$ClientConfig_Path\Spark\install.bat”导入工程依赖包。
3. 单击“File > Open”，选择样例工程的SparkScalaExample文件夹，单击“OK”，工程成功导入。
4. 设置IDEA的文本文件编码格式，解决乱码显示问题。
  在IDEA的菜单栏中，选择“”图标。弹出“Settings”窗口。
  
  在左边导航上选择“File Encodings”，在“IDE encoding”区域，选择参数值为“UTF-8”，单击“Apply”后，单击“OK”，如图6-15所示。
  
  图6-15 File Encodings
（可选）使用IDEA创建一个Spark样例工程。

6.3 开发指引

应用程序实例

假定用户有某个周末网民网购停留时间的日志文本，基于某些业务要求，要求开发Spark应用程序实现如下功能：

统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。
周末两天的日志文件第一列为姓名，第二列为性别，第三列为本次停留时间，单位为分钟，分隔符为“,”。

log1.txt：周六网民停留日志

LiuYang,female,20
YuanJing,male,10
GuoYijun,male,5
CaiXuyu,female,50
Liyuan,male,20
FangBo,female,50
LiuYang,female,20
YuanJing,male,10
GuoYijun,male,50
CaiXuyu,female,50
FangBo,female,60

log2.txt：周日网民停留日志

LiuYang,female,20
YuanJing,male,10
CaiXuyu,female,50
FangBo,female,50
GuoYijun,male,5
CaiXuyu,female,50
Liyuan,male,20
CaiXuyu,female,50
FangBo,female,50
LiuYang,female,20
YuanJing,male,10
FangBo,female,50
GuoYijun,male,50
CaiXuyu,female,50
FangBo,female,60

数据准备

首先需要把原日志文件放置在HDFS系统里。

本地新建两个文本文件，将log1.txt中的内容复制保存到input_data1.txt，将log2.txt中的内容复制保存到input_data2.txt。
在HDFS上建立一个文件夹，“/tmp/input”，并上传input_data1.txt，input_data2.txt到此目录，命令如下：
1. 在linux系统HDFS客户端使用命令“hadoop fs -mkdir /tmp/input（hdfs dfs命令有同样的作用）”，创建对应目录。
2. 在linux系统HDFS客户端使用命令“hadoop fs -put 本地input.txt地址 /tmp/input”，上传数据文件。

代码样例

统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。

Spark的应用程序可使用Java语言或者Scala语言来实现：代码样例请参考6.4 Java代码样例和6.5 Scala代码样例。

6.4 Java代码样例

功能介绍

统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。

主要分为四个部分：

读取原文件数据。
筛选女性网民上网时间数据信息。
汇总每个女性上网总时间。
筛选出停留时间大于两个小时的女性网民信息。

代码样例

下面代码片段仅为演示，具体代码参见com.huawei.bigdata.spark.examples.CollectFemaleInfo类：

    //创建一个配置类SparkConf，然后创建一个SparkContextSparkConf conf = new SparkConf().setAppName("CollectFemaleInfo");JavaSparkContext jsc = new JavaSparkContext(conf);//读取原文件数据,每一行记录转成RDD里面的一个元素JavaRDD<String> data = jsc.textFile(args[0]);//将每条记录的每列切割出来，生成一个TupleJavaRDD<Tuple3<String,String,Integer>> person = data.map(new Function<String,Tuple3<String,String,Integer>>(){private static final long serialVersionUID = -2381522520231963249L;@Overridepublic Tuple3<String, String, Integer> call(String s) throws Exception{//按逗号分割一行数据String[] tokens = s.split(",");//将分割后的三个元素组成一个三元TupleTuple3<String, String, Integer> person = new Tuple3<String, String, Integer>(tokens[0], tokens[1], Integer.parseInt(tokens[2]));return person;}});//使用filter函数筛选出女性网民上网时间数据信息    JavaRDD<Tuple3<String,String,Integer>> female = person.filter(new Function<Tuple3<String,String,Integer>, Boolean>(){private static final long serialVersionUID = -4210609503909770492L;@Overridepublic Boolean call(Tuple3<String, String, Integer> person) throws Exception{//根据第二列性别，筛选出是female的记录Boolean isFemale = person._2().equals("female");return isFemale;}});//汇总每个女性上网总时间JavaPairRDD<String, Integer> females = female.mapToPair(new PairFunction<Tuple3<String, String, Integer>, String, Integer>(){private static final long serialVersionUID = 8313245377656164868L;@Overridepublic Tuple2<String, Integer> call(Tuple3<String, String, Integer> female) throws Exception{//取出姓名和停留时间两列，用于后面按名字求逗留时间的总和Tuple2<String, Integer> femaleAndTime = new  Tuple2<String, Integer>(female._1(), female._3());return femaleAndTime;}}).reduceByKey(new Function2<Integer, Integer, Integer>(){private static final long serialVersionUID = -3271456048413349559L;@Overridepublic Integer call(Integer integer, Integer integer2) throws Exception{//将同一个女性的两次停留时间相加，求和return (integer + integer2);}});//筛选出停留时间大于两个小时的女性网民信息JavaPairRDD<String, Integer> rightFemales = females.filter(new Function<Tuple2<String, Integer>, Boolean>(){private static final long serialVersionUID = -3178168214712105171L;@Overridepublic Boolean call(Tuple2<String, Integer> s) throws Exception{//取出女性用户的总停留时间，并判断是否大于2小时if(s._2() > (2 * 60)){return true;}return false;}});//对符合的female信息进行打印显示for(Tuple2<String, Integer> d: rightFemales.collect()){System.out.println(d._1() + "," + d._2());}

6.5 Scala代码样例

功能介绍

统计日志文件中本周末网购停留总时间超过2个小时的女性网民信息。

主要分为三个部分：

读取女性网民上网时间数据信息组成Spark的RDD数据，通过Spark中类SparkContext中textFile方法实现
从原文件中筛选女性网民上网时间数据信息，通过Spark中RDD中filter方法实现。
汇总每个女性上网时间，并输出时间大于两个小时的女性网民信息，通过Spark中RDD中reduceByKey方法类和filter实现。

代码样例

下面代码片段仅为演示，具体代码参见com.huawei.bigdata.spark.examples.CollectFemaleInfo类：

样例：类CollectMapper

//配置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)
result.foreach(println)

6.6 对外接口

6.6.1 Java API

Spark完整的类及方法参考官方网站的描述： https://spark.apache.org/docs/latest/api/java/index.html

常用接口

Spark主要使用到如下这几个类：

JavaSparkContext：是Spark的对外接口，负责向调用该类的java应用提供Spark的各种功能，如连接Spark集群，创建RDD，累积量和广播量等。它的作用是一个容器。
SparkConf：Spark应用配置类，如设置应用名称，执行模式，executor内存等。
JavaRDD ：用于在java应用中定义JavaRDD的类，功能类似于scala中的“RDD(Resilient Distributed Dataset)”类。
JavaPairRDD：表示key-value形式的JavaRDD。该类提供的方法有groupByKey，reduceByKey等。

Broadcast：广播变量类。广播变量允许保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份拷贝。
StorageLevel：数据存储级别，有内存（MEMORY_ONLY），磁盘（DISK_ONLY），内存+磁盘（MEMORY_AND_DISK）等。

JavaRDD支持两种类型的操作： transformation和action，这两种类型的常用方法如表6-1和表6-2。

表6-1 transformation

方法	说明
map(func)	对调用map的RDD数据集中的每个element都使用func。
filter(func)	对RDD中所有元素调用func，返回f为true的元素。
flatMap(func)	先对RDD所有元素调用func，然后将结果扁平化。
sample(withReplacement,faction,seed)	抽样。
distinct([numTasks])	去除重复元素。
groupByKey(numTasks)	返回(K,Seq[V])，将key相同的value组成一个集合。
reduceByKey(func,[numTasks])	对key相同的value调用func。
sortByKey([ascending],[numTasks])	按照key来进行排序，是升序还是降序，ascending是boolean类型。
join(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,(V,W))的dataset,numTasks为并发的任务数。
cogroup(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,Seq[V],Seq[W])的dataset,numTasks为并发的任务数。
cartesian(otherDataset)	笛卡尔积。

表6-2 action

方法	说明
reduce(func)	对RDD中的元素调用func。
collect()	返回包含RDD中所有元素的一个数组。
count()	返回的是dataset中的element的个数。
first()	返回的是dataset中的第一个元素。
take(n)	返回前n个elements。
takeSample(withReplacement，num，seed)	对dataset随机抽样，返回有num个元素组成的数组。withReplacement表示是否使用replacement。
saveAsTextFile（path）	把dataset写到一个text file中，或者hdfs，或者hdfs支持的文件系统中，spark把每条记录都转换为一行记录，然后写到file中。
saveAsSequenceFile(path)	只能用在key-value对上，然后生成SequenceFile写到本地或者hadoop文件系统。
countByKey()	对RDD中每个元素出现的次数进行统计。
foreach(func)	在数据集的每一个元素上，运行函数func。
countByValue()	对RDD中每个元素出现的次数进行统计。

6.6.2 Scala API

Spark完整的类及方法参考官方网站的描述： https://spark.apache.org/docs/latest/api/scala/index.html#org.apache.spark.package

常用接口

Spark主要使用到如下这几个类：

SparkContext：是Spark的对外接口，负责向调用该类的java应用提供Spark的各种功能，如连接Spark集群，创建RDD等。
SparkConf：Spark应用配置类，如设置应用名称，执行模式，executor内存等。
RDD（Resilient Distributed Dataset）：用于在Spark应用程序中定义RDD的类。
PairRDDFunctions：为key-value对的RDD数据提供运算操作，如groupByKey。
Broadcast：广播变量类。广播变量允许保留一个只读的变量，缓存在每一台机器上，而非每个任务保存一份拷贝。
StorageLevel：数据存储级别，有内存（MEMORY_ONLY），磁盘（DISK_ONLY），内存+磁盘（MEMORY_AND_DISK）等。

RDD上支持两种类型的操作: ：transformation和action，这两种类型的常用方法如表6-3和表6-4。表6-3 transformation

方法	说明
map(func)	对调用map的RDD数据集中的每个element都使用func。
filter(func)	对RDD中所有元素调用func，返回f为true的元素。
flatMap(func)	先对RDD所有元素调用func，然后将结果扁平化。
sample(withReplacement,faction,seed)	抽样。
union(otherDataset)	返回一个新的dataset，包含源dataset和给定dataset的元素的集合。
distinct([numTasks])	去除重复元素。
groupByKey(numTasks)	返回(K,Iterable[V])，将key相同的value组成一个集合。
reduceByKey(func,[numTasks])	对key相同的value调用func。
sortByKey([ascending],[numTasks])	按照key来进行排序，是升序还是降序，ascending是boolean类型。
join(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,(V,W))的dataset,numTasks为并发的任务数。
cogroup(otherDataset,[numTasks])	当有两个KV的dataset(K,V)和(K,W)，返回的是(K,Seq[V],Seq[W])的dataset,numTasks为并发的任务数。
cartesian(otherDataset)	笛卡尔积。

表6-4 action

API	说明
reduce(func)	对RDD中的元素调用func。
collect()	返回包含RDD中所有元素的一个数组。
count()	返回的是dataset中的element的个数。
first()	返回的是dataset中的第一个元素。
take(n)	返回前n个elements。
takeSample(withReplacement，num，seed)	takeSample(withReplacement，num，seed)对dataset随机抽样，返回有num个元素组成的数组。withReplacement表示是否使用replacement。。
saveAsTextFile（path）	把dataset写到一个text file中，或者hdfs，或者hdfs支持的文件系统中，spark把每条记录都转换为一行记录，然后写到file中。
saveAsSequenceFile(path)	只能用在key-value对上，然后生成SequenceFile写到本地或者hadoop文件系统。
countByKey()	对每个key出现的次数做统计。
foreach(func)	在数据集的每一个元素上，运行函数func。
countByValue()	对RDD中每个元素出现的次数进行统计。

6.6.3 Python API

如果您需要使用python语言的客户端运行Spark实例，您可以使用Spark提供的python API。请直接参考官网网站上的详细描述了解其使用： https://spark.apache.org/docs/latest/api/python/index.html

6.6.4 Web UI

操作场景

Spark主要有两个web页面。

SparkUI页面，用于展示正在执行的应用的运行情况。
页面包括了Stages、Storage、Environment和Executors四个部分。
History Server页面，用于展示已经完成的spark应用的运行情况。
页面包括了应用名称、开始时间、结束时间、执行时间、所属用户等信息。单击应用名称，页面将跳转到该应用的SparkUI页面。

说明：

请联系管理员获取具有访问WebUI权限的业务帐号及其密码。

操作步骤

Spark UI打开步骤：
History Server页面打开步骤：

6.6.5 JDBC

JDBC Server是Hive中的HiveServer2的另外一个实现，它底层使用了Spark SQL来处理HQL语句，从而比Hive拥有更高的性能。

JDBC Server默认在安装节点上的22290端口起一个JDBC/ODBC服务，可以通过beeline来连接它，从而执行SQL命令。

Beeline及其他客户端的使用指导，请参见https://cwiki.apache.org/confluence/display/Hive/HiveServer2+Clients。

如果您需要了解支持的Hive特性，请参见Spark官网：http://spark.apache.org/docs/1.1.0/sql-programming-guide.html#compatibility-with-apache-hive

JDBC Server不支持的场景有：

不支持多个客户端同时连接Spark JDBC。
在有索引列的表格中，进行添加列操作后查询会出现错误。例如：
1. 建表：create table part4 (name string, age int, gender string) partitioned by (date string) row format delimited fields terminated by "\t";
2. 导入数据：load data local inpath "/home/sqlfile/people" into table part4 partition (date="2014-10-27");
3. 添加列：alter table part4 add columns (region string comment "where are u live")
4. 查询：select * from part4;