Spark(第二节)Spark角色介绍,Spark提交任务参数解释,分别基于spark-shell、scala、java开发单词计数方法

1300-黄同学

发表文章数:85

热门标签

首页 » 大数据 » 正文

Spark角色介绍

Spark架构图如下:
Spark(第二节)Spark角色介绍,Spark提交任务参数解释,分别基于spark-shell、scala、java开发单词计数方法
Spark架构使用了分布式计算中master-slave模型,master是集群中含有master进程的节点,slave是集群中含有worker进程的节点。

  • Driver Program :运⾏main函数并且新建SparkContext的程序。
  • Application:基于Spark的应用程序,包含了driver程序和集群上的executor。
  • Cluster Manager:指的是在集群上获取资源的外部服务。目前有三种类型
    (1)Standalone: spark原生的资源管理,由Master负责资源的分配
    (2)Apache Mesos:与hadoop MR兼容性良好的一种资源调度框架
    (3)Hadoop Yarn: 主要是指Yarn中的ResourceManager
  • Worker Node: 集群中任何可以运行Application代码的节点,在Standalone模式中指的是通过slaves文件配置的Worker节点,在Spark on Yarn模式下就是NodeManager节点
  • Executor:是在一个worker node上为某应⽤启动的⼀个进程,该进程负责运⾏任务,并且负责将数据存在内存或者磁盘上。每个应⽤都有各自独立的executor。
  • Task :被送到某个executor上的工作单元。

Spark提交任务参数解释

spark任务提交说明

一旦打包好,就可以使用bin/spark-submit脚本启动应用了. 这个脚本负责设置spark使用的classpath和依赖,支持不同类型的集群管理器和发布模式:

bin/spark-submit /
  --class <main-class>
  --master <master-url> /
  --deploy-mode <deploy-mode> /
  --conf <key>=<value> /
  ... # other options
  <application-jar> /
  [application-arguments]

一些基本选项

  1. –class: 你的应用的启动类 (如 org.apache.spark.examples.SparkPi)
  1. –master: 集群的master URL (如 spark://node01:7077)
master参数 参数含义
local 本地以一个worker线程运行(例如非并行的情况).
local[K] 本地以K worker 线程 (理想情况下, K设置为你机器的CPU核数).
local[*] 本地以本机同样核数的线程运行.
spark://HOST:PORT 连接到指定的Spark standalone cluster master. 端口是你的master集群配置的端口,缺省值为7077.
mesos://HOST:PORT 连接到指定的Mesos 集群. Port是你配置的mesos端口, 缺省是5050. 或者如果Mesos使用ZooKeeper,格式为 mesos://zk://…
yarn-client 以client模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.
yarn-cluster 以cluster模式连接到YARN cluster. 集群的位置基于HADOOP_CONF_DIR 变量找到.
  1. –deploy-mode: 是否发布你的驱动到worker节点(cluster) 或者作为一个本地客户端 (client) (default: client)*
  2. –conf: 任意的Spark配置属性, 格式key=value. 如果值包含空格,可以加引号“key=value”. 缺省的Spark配置
  3. application-jar: 打包好的应用jar,包含依赖. 这个URL在集群中全局可见。 比如hdfs:// 共享存储系统, 如果是 file:// path, 那么所有的节点的path都包含同样的jar.
  4. application-arguments: 传给main()方法的参数

参数提交说明

参数 含义
–master MASTER_URL 可以是spark://host:port, mesos://host:port, yarn, yarn-cluster,yarn-client, local
–deploy-mode DEPLOY_MODE Driver程序运行的地方,client或者cluster
–class CLASS_NAME 主类名称,含包名
–name NAME Application名称
–jars JARS Driver依赖的第三方jar包
–py-files PY_FILES 用逗号隔开的放置在Python应用程序PYTHONPATH上的.zip, .egg, .py文件列表
–files FILES 用逗号隔开的要放置在每个executor工作目录的文件列表
–properties-file FILE 设置应用程序属性的文件路径,默认是conf/spark-defaults.conf
–driver-memory MEM Driver程序使用内存大小
–driver-library-path Driver程序的库路径
–driver-class-path Driver程序的类路径
–executor-memory MEM executor内存大小,默认1G
–driver-cores NUM Driver程序的使用CPU个数,仅限于Spark Alone模式
–supervise 失败后是否重启Driver,仅限于Spark Alone模式
–total-executor-cores NUM executor使用的总核数,仅限于Spark Alone、Spark on Mesos模式
–executor-cores NUM 每个executor使用的内核数,默认为1,仅限于Spark on Yarn模式
–queue QUEUE_NAME 提交应用程序给哪个YARN的队列,默认是default队列,仅限于Spark on Yarn模式
–num-executors NUM 启动的executor数量,默认是2个,仅限于Spark on Yarn模式
–archives ARCHIVES 仅限于Spark on Yarn模式

基于spark-shell开发单词计数方法

第一步:准备本地文件

node01服务器执行以下命令准备数据文件

mkdir -p /export/servers/sparkdatas
cd /export/servers/sparkdatas/
vim wordcount.txt

hello me
hello you
hello her

第二步,进入spark-shell

由于是作测试,本地模式进入即可

cd /export/servers/spark-2.2.0-bin-2.6.0-cdh5.14.0/
bin/spark-shell --master local[2]

第三步:开发scala单词统计代码

普通写法

sc.textFile("file:///export/servers/sparkdatas/wordcount.txt").flatMap(x => x.split(" ")).map(x => (x,1)).reduceByKey((x,y) => x + y).collect

简介写法(用_代替匿名函数中只使用一次的参数)

sc.textFile("file:///export/servers/sparkdatas/wordcount.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_ + _).collect

代码说明:
sc:Spark-Shell中已经默认将SparkContext类初始化为对象sc。用户代码如果需要用到,则直接应用sc即可。
textFile:读取数据文件
flatMap:对文件中的每一行数据进行压平切分,这里按照空格分隔。
map:对出现的每一个单词记为1(word,1)
reduceByKey:对相同的单词出现的次数进行累加
collect:触发任务执行,收集结果数据。

基于scala开发spark的单词计数方法

第一步:创建maven工程并导入jar包

pom如下

<properties>
   <scala.version>2.11.8</scala.version>
   <spark.version>2.2.0</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>
   <dependency>
       <groupId>org.apache.hadoop</groupId>
       <artifactId>hadoop-client</artifactId>
       <version>2.7.5</version>
   </dependency>
</dependencies>
<build>
   <sourceDirectory>src/main/scala</sourceDirectory>
   <testSourceDirectory>src/test/scala</testSourceDirectory>
   <plugins>
       <plugin>
           <groupId>org.apache.maven.plugins</groupId>
           <artifactId>maven-compiler-plugin</artifactId>
           <version>3.0</version>
           <configuration>
               <source>1.8</source>
               <target>1.8</target>
               <encoding>UTF-8</encoding>
               <!--    <verbal>true</verbal>-->
           </configuration>
       </plugin>
       <plugin>
           <groupId>net.alchim31.maven</groupId>
           <artifactId>scala-maven-plugin</artifactId>
           <version>3.2.0</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>3.1.1</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>

第二步:创建scala的object并开发scala代码

计划将代码打包发到spark集群上运行,所以将读取文件的路径设为args(0),将保存结果文件的路径设为args(1)

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

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()
  }
}

第三步:准备hdfs上的文件

将之前的wordcount.txt,传到hdfs上

hdfs dfs -mkdir /sparkwordcount
hdfs dfs -put wordcount.txt /sparkwordcount

第四步:代码打包提交到spark集群运行

在IDEA中打包项目,将不含依赖的jar包上传到spark集群的某个机器上。这次使用on-yarn模式来运行。

bin/spark-submit --class wordcount.WordCount /
--master yarn /
--deploy-mode cluster /
--executor-memory 1G /
--total-executor-cores 2 /
/export/servers/spark-word-count.jar /
hdfs://node01:8020/sparkwordcount /
hdfs://node01:8020/sparkwordcount_out

基于java开发spark的单词计数方法

在之前的scala项目中,继续编写java类即可,无需重新创建项目。

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;
import java.util.Iterator;
import java.util.List;

public class WordCountJava
{
    public static void main(String[] args)
    {
        SparkConf conf=new SparkConf().setAppName("word-count").setMaster("local");
        JavaSparkContext context=new JavaSparkContext(conf);
        //读取文件
        JavaRDD<String> text=context.textFile("本地文件路径");
        //对每一行单词进行切分
        JavaRDD<String> flatMapped=text.flatMap(new FlatMapFunction<String, String>()
        {
            @Override
            public Iterator<String> call(String s) throws Exception
            {
                String[] split=s.split(" ");
                Iterator<String> iterator=Arrays.asList(split).iterator();
                return iterator;
            }
        });
        //给每个单词计为 1
        JavaPairRDD<String,Integer> wordAndOne=flatMapped.mapToPair(new PairFunction<String, String, Integer>()
        {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception
            {
                return Tuple2.apply(s,1);
            }
        });
        //相同单词出现的次数累加
        JavaPairRDD<String,Integer> resultJavaPairRDD=wordAndOne.reduceByKey(new Function2<Integer, Integer, Integer>()
        {
            @Override
            public Integer call(Integer integer, Integer integer2) throws Exception
            {
                return integer+integer2;
            }
        });
        //获取结果
        List<Tuple2<String,Integer>> collect=resultJavaPairRDD.collect();
        for (Tuple2<String,Integer> t:collect)
            System.out.println(t.toString());
    }
}

由于Java本身不擅长开发Spark代码,所以这里直接用windows机器上的文件作测试即可,不打包在Spark集群运行,也不生成结果文件了。

标签:
分享到:
赞(0) 打赏

评论 抢沙发

评论前必须登录!

  注册



长按图片转发给朋友

觉得文章有用就打赏一下文章作者

支付宝扫一扫打赏

微信扫一扫打赏

Vieu3.3主题
专业打造轻量级个人企业风格博客主题!专注于前端开发,全站响应式布局自适应模板。

登录

忘记密码 ?

您也可以使用第三方帐号快捷登录

Q Q 登 录
微 博 登 录