第4节 Spark程序：1 - 9

2019-08-26 20:54:41 阅读：139 来源： 互联网

标签：scala 程序 master executor org Spark spark

五、 Spark角色介绍

Spark是基于内存计算的大数据并行计算框架。因为其基于内存计算，比Hadoop中MapReduce计算框架具有更高的实时性，同时保证了高效容错性和可伸缩性。从2009年诞生于AMPLab到现在已经成为Apache顶级开源项目，并成功应用于商业集群中，学习Spark就需要了解其架构。

Spark架构图如下：

Spark架构使用了分布式计算中master-slave模型，master是集群中含有master进程的节点，slave是集群中含有worker进程的节点。

u Driver Program ：运⾏main函数并且新建SparkContext的程序。

u Application：基于Spark的应用程序，包含了driver程序和集群上的executor。

u Cluster Manager：指的是在集群上获取资源的外部服务。目前有三种类型

（1）Standalone: spark原生的资源管理，由Master负责资源的分配

（2）Apache Mesos:与hadoop MR兼容性良好的一种资源调度框架

（3）Hadoop Yarn: 主要是指Yarn中的ResourceManager

u Worker Node：集群中任何可以运行Application代码的节点，在Standalone模式中指的是通过slaves文件配置的Worker节点，在Spark on Yarn模式下就是NodeManager节点

u Executor：是在一个worker node上为某应用启动的⼀个进程，该进程负责运行任务，并且负责将数据存在内存或者磁盘上。每个应用都有各自独立的executor。

u Task ：被送到某个executor上的工作单元。

六、初识Spark程序

6.1 执行第一个spark程序

普通模式提交任务：

bin/spark-submit \

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

--master spark://node1:7077 \

--executor-memory 1G \

--total-executor-cores 2 \

examples/jars/spark-examples_2.11-2.1.3.jar \

该算法是利用蒙特·卡罗算法求圆周率PI，通过计算机模拟大量的随机数，最终会计算出比较精确的π。

高可用模式提交任务：

在高可用模式下，因为涉及到多个Master，所以对于应用程序的提交就有了一点变化，因为应用程序需要知道当前的Master的IP地址和端口。这种HA方案处理这种情况很简单，只需要在SparkContext指向一个Master列表就可以了，

如spark://host1:port1,host2:port2,host3:port3，应用程序会轮询列表，找到活着的Master。

bin/spark-submit \

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

--master spark://node1:7077,node2:7077,node3:7077 \

--executor-memory 1G \

--total-executor-cores 2 \

examples/jars/spark-examples_2.11-2.1.3.jar \

6.2 启动Spark-Shell

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

6.2.1 运行spark-shell --master local[N] 读取本地文件

单机模式：通过本地N个线程跑任务，只运行一个SparkSubmit进程。

（1）需求

读取本地文件，实现文件内的单词计数。本地文件words.txt 内容如下：

hello me

hello you

hello her

（2）运行spark-shell --master local[2]

观察启动的进程：

（3）编写scala代码

sc.textFile("file:///root///words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

代码说明：

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

textFile:读取数据文件

flatMap:对文件中的每一行数据进行压平切分,这里按照空格分隔。

map:对出现的每一个单词记为1（word，1）

reduceByKey:对相同的单词出现的次数进行累加

collect:触发任务执行，收集结果数据。

（4）观察结果：

6.2.2 运行spark-shell --master local[N] 读取HDFS上数据

(1)、整合spark和HDFS，修改配置文件

在spark-env.sh ，添加HADOOP_CONF_DIR配置，指明了hadoop的配置文件后，默认它就是使用的hdfs上的文件

export HADOOP_CONF_DIR=/export/servers/hadoop/etc/hadoop

(2)、再启动启动hdfs，然后重启spark集群

(3)、向hdfs上传一个文件到hdfs://node1:9000/words.txt

(4)、在spark shell中用scala语言编写spark程序

sc.textFile("/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).collect

6.2.3 运行spark-shell 指定具体的master地址

（1）需求:

spark-shell运行时指定具体的master地址，读取HDFS上的数据，做单词计数，然后将结果保存在HDFS上。

（2）执行启动命令：

spark-shell \

--master spark://node1:7077 \

--executor-memory 1g \

--total-executor-cores 2

参数说明：

--master spark://node1:7077 指定Master的地址

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

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

注意：

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

（2）编写scala代码

sc.textFile("/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("/wc")

saveAsTextFile:保存结果数据到文件中

（3）查看hdfs上结果

6.3在IDEA中编写WordCount程序

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

（1）.创建一个项目

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

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

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

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

（6）配置Maven的pom.xml

<properties>

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

        <spark.version>2.1.3</spark.version>

    </properties>

    <dependencies>

        <dependency>

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

            <artifactId>scala-library</artifactId>

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

        </dependency>

        <dependency>

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

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

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

        </dependency>

    </dependencies>

    <build>

        <sourceDirectory>src/main/scala</sourceDirectory>

        <testSourceDirectory>src/test/scala</testSourceDirectory>

        <plugins>

            <plugin>

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

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

                <version>3.2.2</version>

                <executions>

                    <execution>

                        <goals>

                            <goal>compile</goal>

                            <goal>testCompile</goal>

                        </goals>

                        <configuration>

                            <args>

                                <arg>-dependencyfile</arg>

                                <arg>${project.build.directory}/.scala_dependencies</arg>

                            </args>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

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

                            <transformers>

                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">

                                    <mainClass></mainClass>

                                </transformer>

                            </transformers>

                        </configuration>

                    </execution>

                </executions>

            </plugin>

        </plugins>

</build>

（7）添加src/main/scala和src/test/scala，与pom.xml中的配置保持一致

（8）新建一个scala class，类型为Object

（9）.编写spark程序

package cn.test.spark

  

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

  import org.apache.spark.rdd.RDD

  

  

  object WordCount {

  def main(args: Array[String]): Unit = {

    //设置spark的配置文件信息

    val sparkConf: SparkConf = new SparkConf().setAppName("WordCount")

    //构建sparkcontext上下文对象，它是程序的入口,所有计算的源头

    val sc: SparkContext = new SparkContext(sparkConf)

    //读取文件

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

  

    //对文件中每一行单词进行压平切分

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

    //对每一个单词计数为1 转化为(单词，1)

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

    //相同的单词进行汇总 前一个下划线表示累加数据，后一个下划线表示新数据

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

    //保存数据到HDFS

    result.saveAsTextFile(args(1))

    sc.stop()

  }

}

（10）.使用Maven打包：

点击idea右侧的Maven Project选项

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

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

（12）.首先启动hdfs和Spark集群

启动hdfs

/export/servers/hadoop/sbin/start-dfs.sh

启动spark

/export/servers/spark/sbin/start-all.sh

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

spark-submit \

--class cn.test.spark.WordCount \

--master spark://node1:7077 \

--executor-memory 1g \

--total-executor-cores 2 \

/root/spark-1.0-SNAPSHOT.jar \

/words.txt \

/spark_out

这里通过spark-submit提交任务到集群上。用的是spark的Standalone模式

Standalone模式是Spark内部默认实现的一种集群管理模式，这种模式是通过集群中的Master来统一管理资源。

1) 查看Spark的web管理界面

地址： 192.168.200.160:8080

2) 查看HDFS上的结果文件

hdfs dfs -cat /spark_out/part*

(hello,4)

(me,2)

(you,3)

(her,1)

6.4 使用java语言编写spark wordcount程序

package cn.itcast;

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaPairRDD;

import org.apache.spark.api.java.JavaRDD;

import org.apache.spark.api.java.JavaSparkContext;

import org.apache.spark.api.java.function.FlatMapFunction;

import org.apache.spark.api.java.function.Function2;

import org.apache.spark.api.java.function.PairFunction;

import scala.Tuple2;

import java.util.Arrays;

  

  /**

 * java代码实现spark的WordCount

 */

  public class WordCountJava {

    public static void main(String[] args) {

        //todo:1、构建sparkconf,设置配置信息

        SparkConf sparkConf = new SparkConf().setAppName("WordCount_Java").setMaster("local[2]");

        //todo:2、构建java版的sparkContext

        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        //todo:3、读取数据文件

        JavaRDD<String> dataRDD = sc.textFile("d:/data/words1.txt");

        //todo:4、对每一行单词进行切分

        JavaRDD<String> wordsRDD = dataRDD.flatMap(new FlatMapFunction<String, String>() {

            @Override

            public Iterator<String> call(String s) throws Exception {

                String[] words = s.split(" ");

                return Arrays.asList(words).iterator();

            }

        });

        //todo:5、给每个单词计为 1

        // Spark为包含键值对类型的RDD提供了一些专有的操作。这些RDD被称为PairRDD。

        // mapToPair函数会对一个RDD中的每个元素调用f函数，其中原来RDD中的每一个元素都是T类型的，

        // 调用f函数后会进行一定的操作把每个元素都转换成一个<K2,V2>类型的对象,其中Tuple2为多元组

        JavaPairRDD<String, Integer> wordAndOnePairRDD = wordsRDD.mapToPair(new PairFunction<String, String, Integer>() {

            @Override

            public Tuple2<String, Integer> call(String word) throws Exception {

                return new Tuple2<String,Integer>(word, 1);

            }

        });

  

        //todo:6、相同单词出现的次数累加

        JavaPairRDD<String, Integer> resultJavaPairRDD = wordAndOnePairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {

            @Override

            public Integer call(Integer v1, Integer v2) throws Exception {

                return v1 + v2;

            }

        });

  

        //todo:7、反转顺序

        JavaPairRDD<Integer, String> reverseJavaPairRDD = resultJavaPairRDD.mapToPair(new PairFunction<Tuple2<String, Integer>, Integer, String>() {

            @Override

            public Tuple2<Integer, String> call(Tuple2<String, Integer> tuple) throws Exception {

                return new Tuple2<Integer, String>(tuple._2, tuple._1);

            }

        });

  

        //todo:8、把每个单词出现的次数作为key，进行排序，并且在通过mapToPair进行反转顺序后输出

        JavaPairRDD<String, Integer> sortJavaPairRDD = reverseJavaPairRDD.sortByKey(false).mapToPair(new PairFunction<Tuple2<Integer, String>, String, Integer>() {

            @Override

            public Tuple2<String, Integer> call(Tuple2<Integer, String> tuple) throws Exception {

  

                return  new Tuple2<String, Integer>(tuple._2,tuple._1);

                //或者使用tuple.swap() 实现位置互换，生成新的tuple;

            }

        });

  

        //todo:执行输出

        System.out.println(sortJavaPairRDD.collect());

  

        //todo:关闭sparkcontext

        sc.stop();

  

    }

}

标签：scala,程序,master,executor,org,Spark,spark
来源： https://www.cnblogs.com/mediocreWorld/p/11414819.html

本站声明： 1. iCode9 技术分享网（下文简称本站）提供的所有内容，仅供技术学习、探讨和分享；
2. 关于本站的所有留言、评论、转载及引用，纯属内容发起人的个人观点，与本站观点和立场无关；
3. 关于本站的所有言论和文字，纯属内容发起人的个人观点，与本站观点和立场无关；
4. 本站文章均是网友提供，不完全保证技术分享内容的完整性、准确性、时效性、风险性和版权归属；如您发现该文章侵犯了您的权益，可联系我们第一时间进行删除；
5. 本站为非盈利性的个人网站，所有内容不会用来进行牟利，也不会利用任何形式的广告来间接获益，纯粹是为了广大技术爱好者提供技术内容和技术思想的分享性交流网站。

ICode9

第4节 Spark程序：1 - 9

五、 Spark角色介绍

六、 初识Spark程序

6.1 执行第一个spark程序

6.2 启动Spark-Shell

6.2.1 运行spark-shell --master local[N] 读取本地文件

6.2.2 运行spark-shell --master local[N] 读取HDFS上数据

6.2.3 运行spark-shell 指定具体的master地址

6.3在IDEA中编写WordCount程序

6.4 使用java语言编写spark wordcount程序

六、初识Spark程序