Spark入门

1. Spark集群安装

1.1  机器部署

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

1.2  下载Spark安装包

    提前到官网下载相应版本并上传spark-安装包到Linux上

    解压安装包到指定位置

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

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并添加如下配置

export SPARK_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

2. 执行Spark程序

2.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

2.2 启动Spark Shell

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

2.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

    参数说明：

--master spark://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即可.

2.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中

1.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

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>learn.spark</groupId>
    <artifactId>SparkDemo</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <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.6.5</hadoop.version>
        <encoding>UTF-8</encoding>
    </properties>

    <dependencies>
        <!-- 导入scala的依赖 -->
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <!-- 导入spark的依赖 -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <!-- 指定hadoop-client API的版本 -->
        <dependency>
            <groupId>org.apache.hadoop</groupId>
            <artifactId>hadoop-client</artifactId>
            <version>${hadoop.version}</version>
        </dependency>

    </dependencies>

    <build>
        <pluginManagement>
            <plugins>
                <!-- 编译scala的插件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                </plugin>
                <!-- 编译java的插件 -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.5.1</version>
                </plugin>
            </plugins>
        </pluginManagement>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>compile</phase>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>


            <!-- 打jar插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

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

    8.编写spark程序

package spark.scala

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ScalaWordCount {

    def main(args: Array[String]): Unit = {
        // 创建spark配置，设置应用程序名字
        //val conf = new SparkConf().setAppName("ScalaWordCount")
        val conf = new SparkConf().setAppName("ScalaWordCount").setMaster("local[*]")
        // 创建spark执行入口
        val sc = new SparkContext(conf)
        // 指定以后从哪儿读取数据创建RDD（弹性分布式数据集）
        // sc.textFile(args(0)).flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_).sortBy(_._2, false).saveAsTextFile(args(1))

        val line: RDD[String] = sc.textFile(args(0))
        // 切分压平
        val words: RDD[String] = line.flatMap(_.split(" "))
        // 将单词和1组合
        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.使用Maven打包：首先修改pom.xml中的main class

    点击idea右侧的Maven Project选项；点击Lifecycle,选择clean和package，然后点击Run Maven Build

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

    11.首先启动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

    12.使用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 -cat hdfs://node1.edu360.cn:9000/out/part-00000