• Spark系列之Spark启动与基础使用

    title: Spark系列

    第三章 Spark启动与基础使用

    3.1 Spark Shell

    3.1.1 Spark Shell启动

    安装目录的bin目录下面，启动命令：

    spark-shell

    • 1
    $SPARK_HOME/bin/spark-shell \
--master spark://hadoop10:7077 \
--executor-memory 512M \
--total-executor-cores 2

    • 1
    • 2
    • 3
    • 4

    注意上面的 cores 参数，若是 0，那么以后这个 spark shell 中运行的代码是不能执行成功的。千万注意。必要要把 cpu cores 和 memory 设置合理。

    1、executor memory不能超过虚拟机的内存
2、cpu cores不要超过spark集群能够提供的总cpu cores，否则会使用全部。最好不要使用全部。否则其他程序由于没有cpu core可用，就不能正常运行

    • 1
    • 2

    参数说明：

    --master spark://hadoop10:7077 指定Master的地址
--executor-memory 2G 指定每个worker可用内存为2G
--total-executor-cores 2 指定整个集群使用的cpu核数为2个

    • 1
    • 2
    • 3

    注意：

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

    • 1

    Spark-2.x开始/Spark3.x

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

    • 1
    • 2
    • 3

    Spark-1.x

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

    • 1
    • 2
    • 3

    3.1.2 编写WordCount

    在提交WordCount程序之前，先在HDFS集群中的准备一个文件用于做单词统计：

    words.txt内容如下：

    hello huangbo
hello xuzheng
hello wangbaoqiang

    • 1
    • 2
    • 3

    把该文件上传到HDFS文件系统中：

    hadoop fs -mkdir -p /spark/wc/input
hadoop fs -put words.txt /spark/wc/input

    • 1
    • 2

    在 Spark Shell 中提交 WordCount 程序：

    sc.textFile("hdfs://hadoop10/spark/wc/input/words.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).saveAsTextFile("hdfs://hadoop10/spark/wc/output")

    • 1

    查询验证结果：

    hadoop fs -ls hdfs://hadoop10/spark/wc/output
hadoop fs -cat hdfs://hadoop10/spark/wc/output/*

    • 1
    • 2

    说明：

    sc
.textFile("hdfs://hadoop10/spark/wc/input/words.txt")
.flatMap(_.split(""))
.map((_,1))
.reduceByKey(_+_)
.saveAsTextFile("hdfs://hadoop10/spark/wc/output")

1、sc
	是SparkContext对象，该对象时提交spark程序的入口
2、textFile("hdfs://hadoop10/spark/wc/input/words.txt")是从HDFS中读取数据
	底层是通过InputFormat去读取（因为数据在HDFS，从HDFS读取数据的规范就是InputFormat）
	RDD[String] (文件中的一行，就是RDD中的一条数据。)
3、flatMap(_.split(" "))
	先map，再压平，切割压平
	Array(Array("hello", "a"), Array("hello", "b")) =》 Array("hello", "a","hello", "b")
4、map((_,1))
	将单词和1构成元组(word,1)
5、reduceByKey(_+_)
	按照key进行reduce，并将value累加
6、saveAsTextFile("hdfs://hadoop10/spark/wc/output")
	将结果写入到HDFS对应输出目录中


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22

    3.2 Spark Submit

    利用Spark自带的例子程序执行一个求PI（蒙特卡洛算法）的程序：

    [root@hadoop10 bin]# ./spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop10:7077 \
--executor-memory 512m \
--total-executor-cores 2 \
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \
100

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7

    在这里插入图片描述

    运行结果：

    在这里插入图片描述

    理解：

    执行通过/software/spark/bin/spark-shell来运行。
如果不设置master那么默认运行在本机节点；
如果设置 --master spark://hadoop10:7077,那么运行在Spark Standalone模式集群。

    • 1
    • 2
    • 3

    参数说明：

    --class 指定 jar 包的主类

--master 指定jar包提交的模式
详解如下：
1、local
	本地模式，本地运行，可以调试（local 1个线程、local[*]不限线程、local[N] N个线程，理想情况下，N设置为你机器的CPU核数）
2、spark
	提交到Spark Standalone集群，有Master和Worker进程
3、mesos
	将jar包提交到mesos集群，运行在mesos资源管理器框架之上，由mesos负责资源管理，Spark负责任务调度和计算
4、YARN
	将jar包提交到yarn集群，运行在yarn资源管理器框架之上，由yarn负责资源管理，Spark负责任务调度和计算
5、cloud
	比如AWS的EC2，使用这个模式能很方便的访问Amazon的S3，Spark支持多种分布式存储系统，比如HDFS和S3

--deploy-mode 指定jar的运行方式（默认是 client 模式）
详解如下：
1、client 模式
	在提交端生成的JVM会一直等待所有计算过程全部完成才退出，它有两个功能，一个是提交，一个是监控jar包运行（测试环境下使用）
2、cluster 模式
	在提交端生成的JVM在提交jar包后会退出，它只有一个提交功能，然后在某一个 worker上会生成一个Driver的JVM，该JVM执行监控jar包运行的功能，等待所有代码运行完毕退出（生产环境下使用 ）

application.jar 指定你的jar包的地址
arguments       传递给main()方法的参数

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24

    3.2.1 Spark Submit多种运行模式

    1、提交任务到本地运行

    /software/spark/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master local[4] \
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1 \
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \
10

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    2、提交任务到Spark集群运行

    /software/spark/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master spark://hadoop10:7077 \
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1 \
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \
10

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8

    3、提交Yarn集群，使用Yarn-Client模式

    /software/spark/bin/spark-submit \
--class org.apache.spark.examples.SparkPi \
--master yarn \
--deploy-mode client \
--driver-memory 512m \
--executor-memory 512m \
--total-executor-cores 1 \
/software/spark/examples/jars/spark-examples_2.12-3.1.2.jar \
10

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    3.3 修改Spark日志级别

    3.3.1 临时修改

    val sparkContext:SparkContext = new SparkContext(conf)
sparkContext.setLogLevel("WARN")

    • 1
    • 2

    3.3.2 永久修改

    可以通过修改Spark配置文件来Spark日志级别

    以下是详细步骤的：

    第一步：先进入conf目录
[root@hadoop10 conf]$ cd /software/spark/conf

第二步：准备log4j.properties
[root@hadoop10 conf]$ cp log4j.properties.template log4j.properties

第三步：配置日志级别：
把INFO改成你想要的级别：主要有ERROR, WARN, INFO, DEBUG几种


    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9

    在这里插入图片描述

    3.4 Spark的WordCount案例

    3.4.0 spark-shell中的WordCount

    [root@hadoop10 bin]# cd /software/spark/bin/
[root@hadoop10 bin]# pwd
/software/spark/bin
[root@hadoop10 bin]# spark-shell 
2021-11-09 16:57:03,855 WARN util.NativeCodeLoader: Unable to load native-hadoop library for your platform... using builtin-java classes where applicable
Setting default log level to "WARN".
To adjust logging level use sc.setLogLevel(newLevel). For SparkR, use setLogLevel(newLevel).
Spark context Web UI available at http://hadoop10:4040
Spark context available as 'sc' (master = local[*], app id = local-1636448230277).
Spark session available as 'spark'.
Welcome to
      ____              __
     / __/__  ___ _____/ /__
    _\ \/ _ \/ _ `/ __/  '_/
   /___/ .__/\_,_/_/ /_/\_\   version 3.1.2
      /_/
         
Using Scala version 2.12.10 (Java HotSpot(TM) 64-Bit Server VM, Java 1.8.0_202)
Type in expressions to have them evaluated.
Type :help for more information.

scala> sc.textFile("file:///home/data/wordcount.txt").flatMap(_.split(" ")).map((_,1)).reduceByKey(_+_).foreach(println)
(hadoop,1)                                                          (0 + 2) / 2]
(hbase,1)
(hello,3)
(world,1)
                                                                                
scala> 

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28

    3.4.1 Java7版本WordCount

    package com.aa.sparkjava.core.wordcount;

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 org.apache.spark.api.java.function.VoidFunction;
import scala.Tuple2;

import java.util.Arrays;
import java.util.Iterator;

/**
 * @Author AA
 * @Date 2021/11/25 16:46
 * @Project bigdatapre
 * @Package com.aa.sparkjava.core.wordcount
 * WordCountJava java7版本的编写
 */
public class WordCountJava7 {
    public static void main(String[] args){

        //一、参数判断
        if(args.length!=2){
            System.out.println("Usage:JavaWordCount7: Please enter the necessary parameters:");
            System.exit(1);
        }

        //二、编程入口
        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName(WordCountJava7.class.getSimpleName());
        JavaSparkContext jsc = new JavaSparkContext(conf); 

        //三、读取数据，设置输入路径。
        JavaRDD<String> lineRDD = jsc.textFile(args[0]);

        //四、进行逻辑处理，切割压平
        JavaRDD<String> jrdd1 = lineRDD.flatMap(new FlatMapFunction<String, String>() {
            @Override
            public Iterator<String> call(String line) throws Exception {
                return Arrays.asList(line.split(" ")).iterator();
            }
        });

        //五、将四中的单词结果和1组合成元组
        JavaPairRDD<String, Integer> javaPairRDD = jrdd1.mapToPair(new PairFunction<String, String, Integer>() {
            @Override
            public Tuple2<String, Integer> call(String s) throws Exception {
                return new Tuple2<String, Integer>(s, 1);
            }
        });

        //六、 分组聚合  reduceByKey()  (a,b)=>a+b   前两个参数是输入参数类型，第三个参数：返回值的类型
        JavaPairRDD<String, Integer> result = javaPairRDD.reduceByKey(new Function2<Integer, Integer, Integer>() {
            @Override
            public Integer call(Integer v1, Integer v2) throws Exception {
                return v1 + v2;
            }
        });

        //七、保存结果或者打印输出
        //打印输出
        result.foreach(new VoidFunction<Tuple2<String, Integer>>() {
            @Override
            public void call(Tuple2<String, Integer> tuple) throws Exception {
                System.out.println(tuple);
                //System.out.println(tuple._1 + " " + tuple._2);
            }
        });
        //保存结果
        //result.saveAsTextFile(args[1]);
        //八、释放资源
        jsc.close();
    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44
    • 45
    • 46
    • 47
    • 48
    • 49
    • 50
    • 51
    • 52
    • 53
    • 54
    • 55
    • 56
    • 57
    • 58
    • 59
    • 60
    • 61
    • 62
    • 63
    • 64
    • 65
    • 66
    • 67
    • 68
    • 69
    • 70
    • 71
    • 72
    • 73
    • 74
    • 75
    • 76
    • 77
    • 78
    • 79

    3.4.2 Java8 Lambda表达式版本WordCount

    package com.aa.sparkjava.core.wordcount;

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 scala.Tuple2;

import java.util.Arrays;

/**
 * @Author AA
 * @Date 2021/11/26 11:53
 * @Project bigdatapre
 * @Package com.aa.sparkjava.core.wordcount
 * WordCountJava java8版本  使用lambda表达式
 */
public class WordCountJava8 {
    public static void main(String[] args){

        if(args.length != 2){
            System.out.println("Please enter the necessary parameters:Usage Java8 Lambda WordCount");
            System.exit(1);
        }

        SparkConf conf = new SparkConf();
        conf.setMaster("local");
        conf.setAppName(WordCountJava8.class.getSimpleName());
        JavaSparkContext jsc = new JavaSparkContext(conf);
        //1、读取数据
        JavaRDD<String> jrdd = jsc.textFile(args[0]);
        //2、切割压平
        JavaRDD<String> jrdd2 = jrdd.flatMap(t -> Arrays.asList(t.split(" ")).iterator());
        //3、和1组合 
        JavaPairRDD<String, Integer> jprdd = jrdd2.mapToPair(t -> new Tuple2<String, Integer>(t, 1));
        //4、分组聚合
        JavaPairRDD<String, Integer> res = jprdd.reduceByKey((a, b) -> a + b);
        //5、保存输出
        res.saveAsTextFile(args[1]);
        //6、释放资源
        jsc.close();

    }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36
    • 37
    • 38
    • 39
    • 40
    • 41
    • 42
    • 43
    • 44

    3.4.3 Scala版本WordCount

    package com.aa.sparkscala.core.wordcount

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

/**
 * @Author AA
 * @Date 2021/11/26 15:08
 * @Project bigdatapre
 * @Package com.aa.sparkscala.core.wordcount
 * Spark WordCount  Scala版本  使用老的API SparkContext
 */
object WordCountScala {
  def main(args: Array[String]): Unit = {
    //1、创建一个SparkConf对象，并设置程序的名称
    val conf = new SparkConf().setAppName("WordCount").setMaster("local")

    //2、创建一个SparkContext对象
    val sparkContext: SparkContext = new SparkContext(conf)

    //3、读取HDFS上的文件构建一个RDD
    val fileRDD: RDD[String] = sparkContext.textFile("D:\\input\\test1.txt")

    //4、构建一个单词RDD
    val wordAndOneRDD: RDD[(String, Int)] = fileRDD.flatMap(_.split(" ")).map((_, 1))

    //5、进行单词的聚合
    val resultRDD: RDD[(String, Int)] = wordAndOneRDD.reduceByKey(_ + _)

    //6、保存结果
    resultRDD.saveAsTextFile("D:\\output\\wordcountscala1")

    //7、关闭sc
    sparkContext.stop()
  }
}

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10
    • 11
    • 12
    • 13
    • 14
    • 15
    • 16
    • 17
    • 18
    • 19
    • 20
    • 21
    • 22
    • 23
    • 24
    • 25
    • 26
    • 27
    • 28
    • 29
    • 30
    • 31
    • 32
    • 33
    • 34
    • 35
    • 36

    3.4.4 补充API依赖

            <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.12.14</version>
        </dependency>
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>3.1.2</version>
        </dependency>

    • 1
    • 2
    • 3
    • 4
    • 5
    • 6
    • 7
    • 8
    • 9
    • 10


    声明：
            文章中代码及相关语句为自己根据相应理解编写，文章中出现的相关图片为自己实践中的截图和相关技术对应的图片，若有相关异议，请联系删除。感谢。转载请注明出处，感谢。


    By luoyepiaoxue2014

    B站： https://space.bilibili.com/1523287361 点击打开链接
    微博地址： http://weibo.com/luoyepiaoxue2014 点击打开链接

  • 相关阅读:
    容联七陌&百度营销通BCP解决方案，让营销更精准
    【异常----finally和自定义异常】
    C++ 模板同时使用默认参数和偏特化
    Spring Boot + Redis的整合实现
    MISSING COURSE-shell
    【数据库】聚集函数
    安装模块失败?一篇文章教你如何解决掌握镜像安装~
    C语言 不换行了程序输出%1到%100
    缓存中的这7个坑，我把坑惨了！！！
    CAD最常用的快捷键大全来啦
  • 原文地址：https://blog.csdn.net/luoyepiaoxue2014/article/details/128076088