spark入门知识详解_spark讲解

spark 入门

学习目标：

• 了解spark概念
• 知道spark的特点（与hadoop对比）
• 独立实现spark local模式的启动

1.1 spark概述

• 1、什么是spark

  • 基于内存的计算引擎，它的计算速度非常快。但是仅仅只涉及到数据的计算，并没有涉及到数据的存储。

• 2、为什么要学习spark

  MapReduce框架局限性

  • 1，Map结果写磁盘，Reduce写HDFS，多个MR之间通过HDFS交换数据
  • 2，任务调度和启动开销大
  • 3，无法充分利用内存
  • 4，不适合迭代计算（如机器学习、图计算等等），交互式处理（数据挖掘）
  • 5，不适合流式处理（点击日志分析）
  • 6，MapReduce编程不够灵活，仅支持Map和Reduce两种操作

  Hadoop生态圈

  • 批处理：MapReduce、Hive、Pig
  • 流式计算：Storm
  • 交互式计算：Impala、presto

  需要一种灵活的框架可同时进行批处理、流式计算、交互式计算

  • 内存计算引擎，提供cache机制来支持需要反复迭代计算或者多次数据共享，减少数据读取的IO开销
  • DAG引擎，较少多次计算之间中间结果写到HDFS的开销
  • 使用多线程模型来减少task启动开销，shuffle过程中避免不必要的sort操作以及减少磁盘IO

  spark的缺点是：吃内存，不太稳定

• 3、spark特点

  • 1、速度快（比mapreduce在内存中快100倍，在磁盘中快10倍）
    • spark中的job中间结果可以不落地，可以存放在内存中。
    • mapreduce中map和reduce任务都是以进程的方式运行着，而spark中的job是以线程方式运行在进程中。
  • 2、易用性（可以通过java/scala/python/R开发spark应用程序）
  • 3、通用性（可以使用spark sql/spark streaming/mlib/Graphx）
  • 4、兼容性（spark程序可以运行在standalone/yarn/mesos）

1.2 spark启动（local模式）和WordCount(演示)

• 启动pyspark

  • 在$SPARK_HOME/sbin目录下执行

    • ./pyspark

  • 

  • sc = spark.sparkContext
    words = sc.textFile('file:///home/hadoop/tmp/word.txt') \
                .flatMap(lambda line: line.split(" ")) \
                .map(lambda x: (x, 1)) \
                .reduceByKey(lambda a, b: a + b).collect()
    1
    2
    3
    4
    5

  • 输出结果：

    [('python', 2), ('hadoop', 1), ('bc', 1), ('foo', 4), ('test', 2), ('bar', 2), ('quux', 2), ('abc', 2), ('ab', 1), ('you', 1), ('ac', 1), ('bec', 1), ('by', 1), ('see', 1), ('labs', 2), ('me', 1), ('welcome', 1)]
    
    1
    2

e’, 1), (‘labs’, 2), (‘me’, 1), (‘welcome’, 1)]

```
1

spark-core概述

学习目标：

• 知道RDD的概念
• 独立实现RDD的创建

2.1 什么是RDD

• RDD（Resilient Distributed Dataset）叫做弹性分布式数据集，是Spark中最基本的数据抽象，它代表一个不可变、可分区、里面的元素可并行计算的集合.
  • Dataset:一个数据集，简单的理解为集合，用于存放数据的
  • Distributed：它的数据是分布式存储，并且可以做分布式的计算
  • Resilient：弹性的
    • 它表示的是数据可以保存在磁盘，也可以保存在内存中
    • 数据分布式也是弹性的
    • 弹性:并不是指他可以动态扩展，而是容错机制。
      • RDD会在多个节点上存储，就和hdfs的分布式道理是一样的。hdfs文件被切分为多个block存储在各个节点上，而RDD是被切分为多个partition。不同的partition可能在不同的节点上
      • spark读取hdfs的场景下，spark把hdfs的block读到内存就会抽象为spark的partition。
      • spark计算结束，一般会把数据做持久化到hive，hbase，hdfs等等。我们就拿hdfs举例，将RDD持久化到hdfs上，RDD的每个partition就会存成一个文件，如果文件小于128M，就可以理解为一个partition对应hdfs的一个block。反之，如果大于128M，就会被且分为多个block，这样，一个partition就会对应多个block。
  • 不可变
  • 可分区
  • 并行计算

2.2 RDD的创建

• 第一步 创建sparkContext

  • SparkContext, Spark程序的入口. SparkContext代表了和Spark集群的链接, 在Spark集群中通过SparkContext来创建RDD
  • SparkConf 创建SparkContext的时候需要一个SparkConf， 用来传递Spark应用的基本信息

  conf = SparkConf().setAppName(appName).setMaster(master)
  sc = SparkContext(conf=conf)
  1
  2

• 创建RDD

  • 进入pyspark环境

  [hadoop@hadoop000 ~]$ pyspark
  Python 3.5.0 (default, Nov 13 2018, 15:43:53)
  [GCC 4.8.5 20150623 (Red Hat 4.8.5-28)] on linux
  Type "help", "copyright", "credits" or "license" for more information.
  19/03/08 12:19:55 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).
  Welcome to
        ____              __
       / __/__  ___ _____/ /__
      _\ \/ _ \/ _ `/ __/  '_/
     /__ / .__/\_,_/_/ /_/\_\   version 2.3.0
        /_/
  
  Using Python version 3.5.0 (default, Nov 13 2018 15:43:53)
  SparkSession available as 'spark'.
  >>> sc
  <SparkContext master=local[*] appName=PySparkShell>
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16
  17
  18

  • 在spark shell中 已经为我们创建好了 SparkContext 通过sc直接使用
  • 可以在spark UI中看到当前的Spark作业 在浏览器访问当前centos的4040端口

  

  • Parallelized Collections方式创建RDD

    • 调用SparkContext的 parallelize 方法并且传入已有的可迭代对象或者集合

    data = [1, 2, 3, 4, 5]
    distData = sc.parallelize(data)
    1
    2

    >>> data = [1, 2, 3, 4, 5]
    >>> distData = sc.parallelize(data)
    >>> data
    [1, 2, 3, 4, 5]
    >>> distData
    ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:175
    1
    2
    3
    4
    5
    6

    • 在spark ui中观察执行情况

    

    • 在通过parallelize方法创建RDD 的时候可以指定分区数量

    >>> distData = sc.parallelize(data,5)
    >>> distData.reduce(lambda a, b: a + b)
    15
    1
    2
    3

    • 在spark ui中观察执行情况

    

    • Spark将为群集的每个分区（partition）运行一个任务（task）。 通常，可以根据CPU核心数量指定分区数量（每个CPU有2-4个分区）如未指定分区数量，Spark会自动设置分区数。

  • 通过外部数据创建RDD

    • PySpark可以从Hadoop支持的任何存储源创建RDD，包括本地文件系统，HDFS，Cassandra，HBase，Amazon S3等
    • 支持整个目录、多文件、通配符
    • 支持压缩文件

    >>> rdd1 = sc.textFile('file:///home/hadoop/tmp/word.txt')
    >>> rdd1.collect()
    ['foo foo quux labs foo bar quux abc bar see you by test welcome test', 'abc labs foo me python hadoop ab ac bc bec python']
    1
    2
    3

bar quux abc bar see you by test welcome test’, ‘abc labs foo me python hadoop ab ac bc bec python’]
```