1. Spark概述

1.1. 什么是Spark（官网：http://spark.apache.org）

Spark是一种快速、通用、可扩展的大数据分析引擎。目前，Spark生态系统已经包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子项目，Spark是基于内存计算的大数据并行计算框架。Spark基于内存计算，提高了在大数据环境下数据处理的实时性，同时保证了高容错性和高可伸缩性，允许用户将Spark部署在大量廉价硬件之上，形成集群。

1.2. 为什么要学Spark

中间结果输出： Spark是MapReduce的替代方案，而且兼容HDFS、Hive，可融入Hadoop的生态系统，以弥补MapReduce的不足。

1.3. Spark特点

1.3.1. 快

DAG（计算路径的有向无环图）执行引擎，可以通过基于内存来高效处理数据流。

1.3.2. 易用

Spark支持Java、Python和Scala的API，还支持超过80种高级算法，使用户可以快速构建不同的应用。而且Spark支持交互式的Python和Scala的shell，可以非常方便地在这些shell中使用Spark集群来验证解决问题的方法。

1.3.3. 通用

Spark提供了统一的解决方案。Spark可以用于批处理、交互式查询（SparkSQL）、实时流处理（Spark Streaming）、机器学习（Spark MLlib）和图计算（GraphX）。

2. Spark集群安装

2.1. 安装

2.1.1. 机器部署

准备两台以上Linux服务器，安装好JDK

2.1.2. 下载Spark安装包

上传spark-安装包到Linux上

解压安装包到指定位置

tar -zxvf spark-2.1.0-bin-hadoop2.6.tgz -C/usr/local

2.1.3. 配置Spark

进入到Spark安装目录

cd /usr/local/spark-2.1.0-bin-hadoop2.6

进入conf目录并重命名并修改spark-env.sh.template文件

cd conf/

mv spark-env.sh.template spark-env.sh

vi spark-env.sh

在该配置文件中添加如下配置

export JAVA_HOME=/usr/java/jdk1.8.0_111

export SPARK_MASTER_IP=node1.edu360.cn

export SPARK_MASTER_PORT=7077

保存退出

重命名并修改slaves.template文件

mv slaves.template slaves

vi slaves

在该文件中添加子节点所在的位置（Worker节点）

node2.edu360.cn

node3.edu360.cn

node4.edu360.cn

保存退出

将配置好的Spark拷贝到其他节点上

scp -r spark-2.1.0-bin-hadoop2.6/node2.edu360.cn:/usr/local/

scp -r spark-2.1.0-bin-hadoop2.6/node3.edu360.cn:/usr/local/

scp -r spark-2.1.0-bin-hadoop2.6/node4.edu360.cn:/usr/local/

Spark集群配置完毕，目前是1个Master，3个Work，在node1.edu360.cn上启动Spark集群

/usr/local/spark-2.1.0-bin-hadoop2.6/sbin/start-all.sh

启动后执行jps命令，主节点上有Master进程，其他子节点上有Work进行，登录Spark管理界面查看集群状态（主节点）：http://node1.edu360.cn:8080/

到此为止，Spark集群安装完毕，但是有一个很大的问题，那就是Master节点存在单点故障，要解决此问题，就要借助zookeeper，并且启动至少两个Master节点来实现高可靠，配置方式比较简单：

Spark集群规划：node1，node2是Master；node3，node4，node5是Worker

安装配置zk集群，并启动zk集群

停止spark所有服务，修改配置文件spark-env.sh，在该配置文件中删掉SPARK_MASTER_IP并添加如下配置

exportSPARK_DAEMON_JAVA_OPTS="-Dspark.deploy.recoveryMode=ZOOKEEPER-Dspark.deploy.zookeeper.url=zk1,zk2,zk3-Dspark.deploy.zookeeper.dir=/spark"

1.在node1节点上修改slaves配置文件内容指定worker节点

2.在node1上执行sbin/start-all.sh脚本，然后在node2上执行sbin/start-master.sh启动第二个Master

3. 执行Spark程序

3.1. 执行第一个spark程序

/usr/local/spark-2.1.0-bin-hadoop2.6/bin/spark-submit\

--class org.apache.spark.examples.SparkPi \

--master spark://node1.edu360.cn:7077 \

--executor-memory 1G \

--total-executor-cores 2 \

/usr/local/spark-2.1.0-bin-hadoop2.6/lib/spark-examples-2.1.0-hadoop2.6.0.jar\

100

该算法是利用蒙特·卡罗算法求PI

3.2. 启动Spark Shell

spark-shell是Spark自带的交互式Shell程序，方便用户进行交互式编程，用户可以在该命令行下用scala编写spark程序。

3.2.1. 启动spark shell

/usr/local/spark-2.1.0-bin-hadoop2.6/bin/spark-shell\

--master spark://node1.edu360.cn:7077 \

--executor-memory 2g \

--total-executor-cores 2

参数说明：

--masterspark://node1.edu360.cn:7077 指定Master的地址

--executor-memory 2g 指定每个worker可用内存为2G

--total-executor-cores 2 指定整个集群使用的cup核数为2个

注意：

如果启动spark shell时没有指定master地址，但是也可以正常启动spark shell和执行spark shell中的程序，其实是启动了spark的local模式，该模式仅在本机启动一个进程，没有与集群建立联系。

Spark Shell中已经默认将SparkContext类初始化为对象sc。用户代码如果需要用到，则直接应用sc即可

3.2.2. 在spark shell中编写WordCount程序

1.首先启动hdfs

2.向hdfs上传一个文件到hdfs://node1.edu360.cn:9000/words.txt

3.在spark shell中用scala语言编写spark程序

sc.textFile("hdfs://node1.edu360.cn:9000/words.txt").flatMap(_.split(""))

.map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://node1.edu360.cn:9000/out")

4.使用hdfs命令查看结果

hdfs dfs -ls hdfs://node1.edu360.cn:9000/out/p*

说明：

sc是SparkContext对象，该对象时提交spark程序的入口

textFile(hdfs://node1.edu360.cn:9000/words.txt)是hdfs中读取数据

flatMap(_.split(" "))先map在压平

map((_,1))将单词和1构成元组

reduceByKey(_+_)按照key进行reduce，并将value累加

saveAsTextFile("hdfs://node1.edu360.cn:9000/out")将结果写入到hdfs中

3.3. 在IDEA中编写WordCount程序

spark shell仅在测试和验证我们的程序时使用的较多，在生产环境中，通常会在IDE中编制程序，然后打成jar包，然后提交到集群，最常用的是创建一个Maven项目，利用Maven来管理jar包的依赖。

1.创建一个项目

2.选择Maven项目，然后点击next

3.填写maven的GAV，然后点击next

4.填写项目名称，然后点击finish

5.创建好maven项目后，点击Enable Auto-Import

6.配置Maven的pom.xml

<maven.compiler.source>1.8</maven.compiler.source>

<maven.compiler.target>1.8</maven.compiler.target>

<scala.version>2.11.8</scala.version>

<spark.version>2.2.0</spark.version>

<hadoop.version>2.8.0</hadoop.version>

</properties>

<groupId>org.scala-lang</groupId>

<artifactId>scala-library</artifactId>

<version>${scala.version}</version>

</dependency>

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.11</artifactId>

<version>${spark.version}</version>

</dependency>

<groupId>org.apache.hadoop</groupId>

<artifactId>hadoop-client</artifactId>

<version>${hadoop.version}</version>

</dependency>

</dependencies>

<build>

<groupId>net.alchim31.maven</groupId>

<artifactId>scala-maven-plugin</artifactId>

</plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

</plugin>

</plugins>

</pluginManagement>

<groupId>net.alchim31.maven</groupId>

<artifactId>scala-maven-plugin</artifactId>

<id>scala-compile-first</id>

<phase>process-resources</phase>

<goals>

<goal>add-source</goal>

<goal>compile</goal>

</goals>

</execution>

<id>scala-test-compile</id>

<phase>process-test-resources</phase>

<goals>

<goal>testCompile</goal>

</goals>

</execution>

</executions>

</plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

<phase>compile</phase>

<goals>

<goal>compile</goal>

</goals>

</execution>

</executions>

</plugin>

<groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-shade-plugin</artifactId>

<phase>package</phase>

<goals>

<goal>shade</goal>

</goals>

</excludes>

</filter>

</filters>

</configuration>

</execution>

</executions>

</plugin>

</plugins>

</build>

7.新建一个scala class，类型为Object

8.编写spark程序

val conf =new SparkConf().setAppName("WordCount").setMaster("local[4]")

//sparkContext是spark程序执行的入口

val sc =new SparkContext(conf)

//使用SparkContext创建RDD

//告诉spark以后从哪里读取数据（不会立即读取数据，是lazy）

//sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).sortBy(_._2, false).saveAsTextFile(args(1))

val lines: RDD[String] = sc.textFile(args(0))

//切分压平数据

val words: RDD[String] = lines.flatMap(_.split(" "))

//将单词和一组合在一起

val wordAndOne: RDD[(String, Int)] = words.map((_,1))

//按key进行聚合

val reduced: RDD[(String, Int)] = wordAndOne.reduceByKey(_+_)

//排序

val sorted: RDD[(String, Int)] = reduced.sortBy(_._2,false)

//产生结果（将数据保存到hdfs中）

sorted.saveAsTextFile(args(1))

//释放资源

sc.stop()

9.

10.使用Maven打包：首先修改pom.xml中的main class

点击idea右侧的MavenProject选项

点击Lifecycle,选择clean和package，然后点击Run Maven Build

11.选择编译成功的jar包，并将该jar上传到Spark集群中的某个节点上

12.首先启动hdfs和Spark集群

启动hdfs

/usr/local/hadoop-2.6.5/sbin/start-dfs.sh

启动spark

/usr/local/spark-2.1.0-bin-hadoop2.6/sbin/start-all.sh

13.使用spark-submit命令提交Spark应用（注意参数的顺序）

/usr/local/spark-2.1.0-bin-hadoop2.6/bin/spark-submit\

--class cn.itcast.spark.WordCount \

--master spark://node1.edu360.cn:7077 \

--executor-memory 2G \

--total-executor-cores 4 \

/root/spark-mvn-1.0-SNAPSHOT.jar \

hdfs://node1.edu360.cn:9000/words.txt \

hdfs://node1.edu360.cn:9000/out

查看程序执行结果

hdfs dfs -cathdfs://node1.edu360.cn:9000/out/part-00000

大数据-spark概述1. Spark概述2. Spark集群安装3. 执行Spark程序

1. Spark概述

1.1. 什么是Spark（官网：http://spark.apache.org）

1.2. 为什么要学Spark

1.3. Spark特点

1.3.1. 快

1.3.2. 易用

1.3.3. 通用

2. Spark集群安装

2.1. 安装

2.1.1. 机器部署

2.1.2. 下载Spark安装包

2.1.3. 配置Spark

3. 执行Spark程序

3.1. 执行第一个spark程序

3.2. 启动Spark Shell

3.2.1. 启动spark shell

3.2.2. 在spark shell中编写WordCount程序

3.3. 在IDEA中编写WordCount程序

继续阅读

【51CTO学院三周年】自学路上的伴侣

在线教育巨头多邻国Duolingo入华一周年，中国市场马力全开

【分类算法】什么是分类算法定义分类与聚类分类过程方法

申请评分模型拒绝推断（RI）方法申请评分模型拒绝推断（RI）方法

Sql优化一：sql语句优化

Nacos 2.0 升级前后性能对比压测

尚硅谷—韩顺平—图解 Java设计模式（结构型）（55～）

Storm编译打包过程中遇到的一些问题及解决方法

MapReduce的几个企业级经典面试案例MapReduce的几个企业级经典面试案例

9.spark Core 进阶2--Cashe

大数据排错SparkSpark集群启动时候，JAVA_HOME is not sethadoop集群，某台服务器jps无任何输出IDEAkafkahadoopspark sqlfile permissionsIDEA本地测试 - OutOfMemoryError: GC overhead limit exceededhdfs负载均衡

浅谈企业活动中进行数据分析的重要性

Ambari介绍和架构原理

spark/scala关于【资源文件】加载方法概述外部文件加载方案测试资源文件打包入jar包中小结

NOSQL安全攻击

win10本地scala和spark安装安装scala安装spark