spark學習

• 2021 年 7 月 16 日
• 筆記
• spark

Spark概述

Spark定義

spark是一種基於記憶體的快速、通用、可擴展S的大數據分析計算引擎

Spark Core 中提供了Spark最基礎核心的功能 Spark SQL是Spark用來操作結構化數據的組件 Spark Streaming是Spark平台上針對實時數據進行流式計算的組件，提供了豐富的處理數據流的ApI

Spark和Hadoop

Spark和Hadoop 的根本差異是多個作業之間的數據通訊問題 : Spark 多個作業之間數據通訊是基於記憶體，而 Hadoop 是基於磁碟。

Hadoop mapreduce在並行運行的蘇劇可復用場景存在很多計算效率問題，Spark在Hadoop的MapReduce計算框架基礎上而誕生，大大加快了數據分析，數據挖掘和讀寫速度，將計算單元縮小到更適合併行計算和重複使用的RDD計算模型

Spark是一個分散式數據快速分析項目，核心技術就是彈性分散式數據集RDD，提供了比MR豐富的模型，可以快速在記憶體中對數據集進行多次迭代，來支援複雜的數據挖掘演算法和影像計算演算法

Spark只有在shuffle的時候會將數據寫入磁碟當中，而Hadoop的MR作業需要依賴磁碟交互，Spark的快取機制比HDFS的更高效

Spark核心模組

Spark Core

Spark Core 中提供了 Spark 最基礎與最核心的功能，Spark 其他的功能如：Spark SQL，

Spark Streaming，GraphX, MLlib 都是在 Spark Core 的基礎上進行擴展的

➢ Spark SQL

Spark SQL 是 Spark 用來操作結構化數據的組件。通過 Spark SQL，用戶可以使用 SQL

或者 Apache Hive 版本的 SQL 方言（HQL）來查詢數據。

➢ Spark Streaming

Spark Streaming 是 Spark 平台上針對實時數據進行流式計算的組件，提供了豐富的處理

數據流的 API。

➢ Spark MLlib

MLlib 是 Spark 提供的一個機器學習演算法庫。MLlib 不僅提供了模型評估、數據導入等

額外的功能，還提供了一些更底層的機器學習原語。

➢ Spark GraphX

GraphX 是 Spark 面向圖計算提供的框架與演算法庫。

Spark運行環境

本地模式

將Hadoop2.7.6的壓縮包解壓到指定目錄文件夾，在bin文件下添加winutils.exe

在idea添加依賴,Scala相應的插件也要下載，下載後重啟idea就好了

 <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>2.11.12</version>
        </dependency>
​
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-compiler</artifactId>
            <version>2.11.12</version>
        </dependency>
​
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-reflect</artifactId>
            <version>2.11.12</version>
        </dependency>
        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.40</version>
        </dependency>
        <!-- //mvnrepository.com/artifact/org.apache.spark/spark-core -->
        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.11</artifactId>
            <version>2.4.5</version>
        </dependency>
    </dependencies>
    <build>
        <plugins>
            <!-- Java Compiler -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.1</version>
                <configuration>
                    <source>1.8</source>
                    <target>1.8</target>
                </configuration>
            </plugin>
​
            <!-- Scala Compiler -->
            <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
​
        </plugins>
    </build>
​

簡單的測試一下本地運行是否成功

package sparkcore
​
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
​
object Demo01WordCount {
  def main(args: Array[String]): Unit = {
    /**
      * 1、創建是spark的環境
      */
​
    //spark配置文件對象
    val conf = new SparkConf()
​
    //任務名
    conf.setAppName("wc")
​
    //指定運行方法local：本地運行
    conf.setMaster("local")
​
    //創建spark上下文對象，spark入口
    val sc = new SparkContext(conf)
​
    /**
      * RDD:彈性的分散式數據集
      * 可以理解為一個集合
      */
    //1、讀取數據
    val linesRDD: RDD[String] = sc.textFile("data/words.txt")
​
    //2、將單詞切分出來
    val wordsRDD: RDD[String] = linesRDD.flatMap(line => line.split(","))
​
    //3、按照單詞進行分組
    val groupRDD: RDD[(String, Iterable[String])] = wordsRDD.groupBy(word => word)
​
    //4、統計單詞的數量
    val countRDD: RDD[(String, Int)] = groupRDD.map {
      case (word: String, ws: Iterable[String]) =>
        //計算單詞的數量
        val count: Int = ws.size
        (word, count)
    }
​
    //5、整理數據
    val resultRDD: RDD[String] =countRDD.map {
      case (word: String, count: Int) =>
        word + "\t" + count
    }
    //5、保存數據
    resultRDD.saveAsTextFile("data/wc")
  }
​
}
​

在說RDD之前先複習一遍mapreduce