Spark入門介紹

2021-02-15 大數據挖掘與機器學習

本篇是介紹Spark的入門系列文章，希望能幫你初窺Spark的大門。

一 Spark概述

1.1 什麼是Spark？

Spark是一種基於內存的快速，通用，可擴展的大數據計算引擎框架。

由於MapReduce計算框架在Map和Reduce階段存在大量的shuffle操作以及IO操作，導致MapReduce的效率一直被很多同學吐槽，而Spark計算框架一定程度下減了shuffle及寫入磁碟操作，因此速度相比較快，因此今天介紹基於Hadoop的Spark計算框架；

先來看一張Hadoop與Spark框架關係圖，Hadoop與Spark關係一目了然，當然並不是說Spark可以代替MapReduce框架，只能說在某一方面上優勝MapReduce，但仍然無法代替MapReduce計算框架。

MapReduce計算框架會存在一定的局限性

1. 基於HDFS文件系統，經過Map計算及Reduce聚合結果保存到磁碟中，對於數據挖掘及數據分析來說，需要對數據進行多次操作，在一定程度上存在局限性；

2. MapReduce在Map及Reduce階段需要大量IO及工作節點移動文件需要的帶寬資源，因此性能上也存在一定的局限性；

3. MapReduce和Hadoop緊密耦合在一起，無法動態替換。

在大家急需一種新的計算引擎出現的情況下，Spark應運而生，不過Spark真正與Hadoop產生聯繫，還需要Yarn幫忙。

 

在2013年10月發布Hadoop2.X之前，Hadoop1.x是沒有Yarn存在的，存儲和計算緊密耦合，並且負責調度任務和資源的JobTracker苦不堪言，所有工作都堆在了它身上。在2.X之後，Yarn作為資源管理器閃亮登場，通過ResourceManager和NodeManager將任務調度和資源調度的職責分開，再通過ApplicationMaster和Cotainer，實現ResourceManager和NodeManager的解耦合。

於是，Hadoop1.x中的hdfs+mr，變成了Hadoop2.x中的ResourceManager+ApplicationMaster+NodeManager，實現了存儲和計算的解耦合，並且實現了Container（可以理解為電腦上裝的虛擬機）中計算引擎的可插拔替換（MR，Spark，Tez，Presto等），至此，Spark可以一展身手了。

二、Spark的幾大組件介紹

主要四大組件：

Spark Core：實現了Spark的基本功能，包含任務調度、內存管理、錯誤恢復、與存儲系統交互等模塊。SparkCore中還包含了對彈性分布式數據集(Resilient Distributed DataSet，簡稱RDD)的API定義。

Spark SQL：是Spark用來操作結構化數據的程序包。通過Spark SQL，我們可以使用 SQL或者Apache Hive版本的SQL方言(HQL)來查詢數據。Spark SQL支持多種數據源，比如Hive表、Parquet以及JSON等。

Spark Streaming：是Spark提供的對實時數據進行流式計算的組件。提供了用來操作數據流的API，並且與Spark Core中的 RDD API高度對應。

Spark ML：提供常見的機器學習(ML)功能的程序庫。包括分類、回歸、聚類、協同過濾等，還提供了模型評估、數據 導入等額外的支持功能。

集群管理器：Spark 設計為可以高效地在一個計算節點到數千個計算節點之間伸縮計算。為了實現這樣的要求，同時獲得最大靈活性，Spark支持在各種集群管理器(Cluster Manager)上運行，包括Hadoop YARN、Apache Mesos，甚至可以在Spark自帶的簡易調度器上運行。

 

 三、Spark運行模式

Spark的運行模式模式，一般分為Local模式，Standalone模式及Yarn模式（重點），Mesos模式（此模式國內基本不用）。

Local模式：運行在一臺計算機上的模式，通常用於本機練手和測試；

Standalone模式：構建一個由Master+Slaver構成的Spark集群；

Yarn模式：Spark客戶端直接連接Yarn，不需要額外構建Spark集群。有yarn-client和yarn-cluster兩種模式，主要區別在於：Driver程序的運行節點。

yarn-client：Driver程序運行在客戶端，適用於交互、調試，希望立即看到app的輸出yarn-cluster：Driver程序運行在由RM（ResourceManager）啟動的AP（APPMaster）適用於生產環境。

Mesos模式（了解）：Spark客戶端直接連接Mesos；不需要額外構建Spark集群。國內應用比較少，更多的是運用yarn調度。

四 、Spark及相應環境安裝

4.1 Mac：

下載安裝jdk1.8並配置環境變量，下載scala的壓縮包後解壓（我使用的是scala-2.11.12），；以及spark-2.3.1-bin-hadoop2.7壓縮包解壓。

最後記得vim .bash_profile配置三者的環境：

exportSCALA_HOME=/User/xxx/xxx/scala-2.11.12

export PATH=$PATH:$SCALA_HOME/bin

SPARK_LOCAL_IP=127.0.0.1

export JAVA_HOME=」/Library/Java/JavaVirtualMachines/jdk1.8_0_221.jdk/Contents/Home」

配置後更新環境變量source .bash_profile

若能在console輸入java -version返回版本則為成功；

若能在console輸出scala -version返回版本則為成功

操作完之後，在spark文件夾下，輸入./bin/spark-shell若成功列印出：

Welcome to

      ____              __

     / __/__ ___ _____/ /__

    _\ \/ _ \/ _ `/ __/  '_/

   /___/ .__/\_,_/_/ /_/\_\   version 2.3.1

      /_/

 

則表示成功。

 

4.2 Windows：

安裝jdk1.8並配置環境變量，若能在cmd輸入java -version返回版本則為成功；解壓scala-2.11.12並配置環境變量，若能在cmd輸出scala -version返回版本則為成功；解壓spark-2.3.2-bin-hadoop2.7並配置環境變量，若能在cmd輸入spark-shell出現welcome語則為成功；解壓hadoop-2.7.7並配置環境變量（winutils.exe已放置hadoop-2.7.7\bin目錄下）

 

4.3 用IDEA跑WordCount，此案例等同於其他語言的Hello World。

 

打開IDEA，新建scala項目，選擇之前安裝的jdk和scala版本，進入項目後，右擊項目，選擇Open Module Settings，選擇Libraries，點擊加號，java，把我們解壓縮的spark-2.3.2-bin-hadoop2.7的 jars目錄下所有jar包選中，點擊ok，給依賴包重取個名字，再次點擊ok，這樣所有依賴的jar包就都有了。

 

 

在你的項目下新建input文件夾，裡面可以新建一個或兩個txt文件，裡面寫上hello world，hello spark等文字，再新建一個WordCount的object文件，輸入以下內容（我們的程序會自動讀取input文件夾裡的文件）：

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.rdd.RDD
object WordCount {
  def main(args: Array[String]): Unit = {
    // local模式
    // 創建SparkConf對象
   // 設定Spark計算框架的運行環境
    // app id
    val config:SparkConf = new SparkConf().setMaster("local[2]").setAppName("WordCount")
    // 創建spark上下文對象
    val sc = new SparkContext(config)
    // 讀取文件，將文件內容一行行的讀取出來
    val unit: RDD[String] = sc.textFile("input")
    // 將一行行的數據分解為一個個的單詞
    val unit1: RDD[String] = unit.flatMap(_.split(" "))
    // 為了統計方便，將單詞數據進行結構的轉換
    val unit2: RDD[(String, Int)] = unit1.map((_,1))
    // 對轉換結構後的數據進行分組聚合
    val unit3: RDD[(String, Int)]  = unit2.reduceByKey(_+_)
    // 收集結果並循環列印
    unit3.collect().foreach(println)
  }
}

右擊run，如果能正確地在控制臺列印出結果，那麼恭喜你，你的第一個Spark程序已經運行成功了！

 

五、Spark中的rdd和算子

RDD（Resilient Distributed Dataset）叫做彈性分布式數據集，是Spark中最基本的數據抽象。代碼中是一個抽象類，它代表一個不可變、可分區、裡面的元素可並行計算的集合。

在Spark中，RDD被表示為對象，主要操作方法有轉換及動作兩大方面。

通過對象上的方法調用來對RDD進行轉換。經過一系列的transformations定義RDD之後，就可以調用actions觸發RDD的計算，action可以是向應用程式返回結果(count, collect等)，或者是向存儲系統保存數據(saveAsTextFile等)。在Spark中，只有遇到action，才會執行RDD的計算(即延遲計算)，這樣在運行時可以通過管道的方式傳輸多個轉換。

 

5.1 RDD轉換（transformations）

整體上分為Value類型和Key-Value類型；

a、Value類型主要有:

map(func);

mapPartitions(func);

mapPartitionsWithIndex(func);

flatMap(func);

glom;

groupBy(func);

filter(func);

sample(withReplacement,fraction,seed);

distinct([numTasks]));

coalesce(numPartitions);

sortBy(func,[ascending], [numTasks]);

pipe(command, [envVars])

 

b、雙Value類型主要有:

union(otherDataset);

subtract (otherDataset);

intersection(otherDataset);

cartesian(otherDataset);

zip(otherDataset)

 

c、Key-Value類型主要有:

partitionBy;

groupByKey;

reduceByKey(func,[numTasks]);

aggregateByKey;

foldByKey;

combineByKey[C];

sortByKey([ascending], [numTasks]);

mapValues;

join(otherDataset, [numTasks]);

cogroup(otherDataset, [numTasks])

 

5.2 RDD的執行(actions)，主要有:

reduce(func);

collect();

count();

first();

take(n); 

takeOrdered(n);

aggregate;

fold(num)(func);

saveAsTextFile(path);

saveAsSequenceFile(path);

saveAsObjectFile(path);

countByKey();

foreach(func)

 

每個算子都有其不同的用法，由於篇幅關係，關於RDD的內容就先聊到這裡，相關算子介紹請看下期~

相關知識：

centos7 分布式集群hadoop與hive安裝

微信公眾號：

相關焦點

  • 2小時入門Spark之MLlib
    一，MLlib基本介紹MLlib是Spark的機器學習庫，包括以下主要功能。實用工具：線性代數，統計，數據處理等工具特徵工程：特徵提取，特徵轉換，特徵選擇常用算法：分類，回歸，聚類，協同過濾，降維模型優化：模型評估，參數優化。MLlib庫包括兩個不同的部分。
  • PySpark 之Spark DataFrame入門
    創建Spark DataFrame其實我的文章裡已經介紹了幾張創建DataFrame的方法，下面介紹下其它方式，Spark內置了多種數據格式的支持(jso, parquet,  jdbc, orc, libsvm, csv, text 等)SQL （直接通過SQL
  • 大數據分析工程師入門9-Spark SQL
    本文為《大數據分析師入門課程》系列的第9篇，在本系列的第8篇-Spark基礎中，已經對Spark做了一個入門介紹，在此基礎上本篇拎出Spark SQL，主要站在使用者的角度來進行講解，需要注意的是本文中的例子的代碼均使用Scala語言。
  • Spark入門必讀:核心概念介紹及常用RDD操作
    核心概念介紹Spark架構示意圖如圖2-1所示，下面將分別介紹各核心組件。Client：客戶端進程，負責提交作業。RDD（Resilient Distributed Dataset）：彈性分布式數據集，可以理解為一種只讀的分布式多分區的數組，Spark計算操作都是基於RDD進行的，下面會有詳細介紹。DAG（Directed Acyclic Graph）：有向無環圖。
  • Apache Spark大數據分析入門(一)
    鑑於此，本文通過動手實戰操作演示帶領大家快速地入門學習Spark。本文是Apache Spark入門系列教程（共四部分）的第一部分。全文共包括四個部分：第一部分：Spark入門，介紹如何使用Shell及RDDs第二部分：介紹Spark SQL、Dataframes及如何結合Spark與Cassandra一起使用第三部分：介紹Spark MLlib和Spark Streaming第四部分：介紹Spark Graphx圖計算本篇講解的便是第一部分
  • 大數據入門:Spark RDD、DataFrame、DataSet
    今天的大數據入門分享，我們就主要來講講Spark RDD、DataFrame、DataSet。RDD，作為Spark的核心數據抽象，是Spark當中不可或缺的存在，而在SparkSQL中，Spark為我們提供了兩個新的抽象，分別是DataFrame和DataSet。
  • Spark 1.6.0 新手快速入門
    本文簡單介紹了Spark的使用方式。首先介紹Spark的交互界面的API使用，然後介紹如何使用Java、Scala以及Python編寫Spark應用。詳細的介紹請閱讀Spark Programming Guide。 在按照本文進行操作之前，請確保已安裝Spark。
  • 2小時入門SparkSQL編程
    本節將介紹SparkSQL編程基本概念和基本用法。不同於RDD編程的命令式編程範式，SparkSQL編程是一種聲明式編程範式，我們可以通過SQL語句或者調用DataFrame的相關API描述我們想要實現的操作。
  • 大數據學習路線教程圖,如何快速入門Spark
    Spark是大數據中重要的框架之一，下面為大家分享如何快速入門spark。Apache Spark是在大數據工業界裡用的最多的基於內存的技術框架，尤其是RDD的特性以及應用，對幫助理解Spark和任務提交的流程以及緩存機制。通過以上教程可以讓大家掌握Spark的環境搭建，任務調度流程，以及RDD代碼的應用。
  • Spark—15分鐘教程
    無論你是想快速入門介紹sparksql，還是急於編寫你的程序，還是像我一樣需要一份備忘單，我相信你會發現這篇文章很有用。這篇文章的目的是介紹sparksql的所有主要函數/特性，在片段中，你將始終看到原始的SQL查詢及其在PySpark中的翻譯。
  • 『 Spark 』2. spark 基本概念解析
    過程中的理解記錄 ＋ 對參考文章中的一些理解 ＋ 個人實踐spark過程中的一些心得而來。寫這樣一個系列僅僅是為了梳理個人學習spark的筆記記錄，並非為了做什麼教程，所以一切以個人理解梳理為主，沒有必要的細節就不會記錄了。若想深入了解，最好閱讀參考文章和官方文檔。其次，本系列是基於目前最新的 spark 1.6.0 系列開始的，spark 目前的更新速度很快，記錄一下版本好還是必要的。1.
  • spark streaming流處理入門乾貨,傾力奉獻
    02spark streaming概述Spark Streaming 提供一個對於流數據的抽象 DStream。spark也同樣支持滑動窗口操作。05小結spark streaming作為spark中的流處理組件，把連續的流數據按照時間間隔劃分為一個個數據塊，然後對每個數據塊分別進行批處理
  • Spark中的RDD介紹
    常見的RDD操作有哪些（包括RDD的分類）RDD的依賴關係（DAG）RDD依賴關係的劃分（stage）RDD為什麼是Spark的核心概念Spark建立在統一抽象的RDD之上，使得Spark可以很容易擴展，比如 Spark Streaming、Spark SQL、Machine Learning、Graph都是在spark
  • Spark 2.0.0 支持 Java 8 Lambda
    lt;artifactId>spark-core</artifactId>        <version>2.0.0</version></dependency>入門指南import static spark.Spark
  • Spark機器學習.pdf
    PDF 獲取方式關注微信公眾號 datayx  然後回復 spark  即可獲取。AI項目體驗地址 https://loveai.tech《Spark機器學習》內容提要彭特裡思著的《Spark機器學習》每章都設計了 案例研究，以機器學習算法為主線，結合實例探討了 spark的實際應用。
  • 2小時入門Spark之RDD編程
    本節將介紹RDD數據結構的常用函數。包括如下內容:這些函數中，我最常用的是如下15個函數，需要認真掌握其用法。import findspark#指定spark_home為剛才的解壓路徑,指定python路徑spark_home = "/Users/liangyun/ProgramFiles/spark-3.0.1-bin-hadoop3.2"python_path
  • 當MongoDB遇見Spark
    所以在讀取上更加快MongoDB支持的增刪改功能比HDFS更加易於修改寫入後的數據HDFS的響應級別為分鐘, 而MongoDB通常是毫秒級別如果現有資料庫已經是MongoDB的話, 那就不用再轉存一份到HDFS上了可以利用MongoDB強大的Aggregate做數據的篩選或預處理MongoDB Spark Connector介紹
  • 『 Spark 』13. Spark 2.0 Release Notes 中文版
    過程中的理解記錄 ＋ 對參考文章中的一些理解 ＋ 個人實踐spark過程中的一些心得而來。寫這樣一個系列僅僅是為了梳理個人學習spark的筆記記錄，所以一切以能夠理解為主，沒有必要的細節就不會記錄了，而且文中有時候會出現英文原版文檔，只要不影響理解，都不翻譯了。若想深入了解，最好閱讀參考文章和官方文檔。其次，本系列是基於目前最新的 spark 1.6.0 系列開始的，spark 目前的更新速度很快，記錄一下版本號還是必要的。
  • 『 Spark 』9. 搭建 IPython + Notebook + Spark 開發環境
    過程中的理解記錄 ＋ 對參考文章中的一些理解 ＋ 個人實踐spark過程中的一些心得而來。寫這樣一個系列僅僅是為了梳理個人學習spark的筆記記錄，所以一切以能夠理解為主，沒有必要的細節就不會記錄了，而且文中有時候會出現英文原版文檔，只要不影響理解，都不翻譯了。若想深入了解，最好閱讀參考文章和官方文檔。其次，本系列是基於目前最新的 spark 1.6.0 系列開始的，spark 目前的更新速度很快，記錄一下版本號還是必要的。
  • Spark 2.0系列之SparkSession詳解
    的各項功能，用戶不但可以使用DataFrame和Dataset的各種API，學習Spark的難度也會大大降低。SparkSession的功能首先，我們從一個Spark應用案例入手：SparkSessionZipsExample可以從JSON文件中讀取郵政編碼，通過DataFrame API進行分析，同時還能夠使用Spark SQL語句實施查詢。