Spark基礎和RDD

spark

1. Spark的四大特性

1. 速度快

• spark比mapreduce快的兩個緣由

  • 1. 基於內存

  1. mapreduce任務後期在計算的是時候，每個job的輸出結果都會落地到磁盤，後續有其餘的job要依賴於前面job的輸出結果，這個時候就須要進行大量的磁盤io操做，性能較低
  
  2. spark任務後期在進行計算的時候，job的結果是能夠保存在內存中的，後面有其餘的job須要以言語前面job的輸出結果，這個時候能夠直接從內存中讀取，避免了磁盤io操做，性能比較高
  
  spark程序和mapreduce程序都會產生shuffle階段，在shuffle階段中他們產生的數據都會保留在磁盤中

  • 1. 進程與線程

  1 mapreduce任務以進程的方式運行在yarn集羣中，好比說有100個mapTask，一個task就須要一個進程，這些task要運行就須要開啓100個進程
  
    2 spark任務以線程的方式運行在進程中,好比說有100個task，則一個task就對應一個線程

  1. 易用性
  2. 能夠用java、scala、python、R等不一樣的語言來快速編寫spark程序
  3. 通用性
  4. 兼容性
     1. spark程序有多種運行模式
        • standAlone
          • spark自帶的獨立運行模式，整個任務的資源分配由spark集羣的的Master來負責
        • yarn
          • 能夠把spark程序提交到yarn上運行，整個任務的資源分配由yarn中的ResourceManager負責
        • mesos
          • apache開源的一個相似於yarn的資源調度平臺

2. spark集羣架構

​

• Driver
  • 他會執行客戶端寫好的main方法，構建一個SparkContext對象（該對象是全部spark程序的執行入口）
• Application
  • 是一個spark的應用程序，包含了客戶端的代碼和任務運行的資源信息
• ClusterManager
  • 給程序提供計算資源的外部服務
    • standAlone
      • spark自帶的集羣模式，整個任務的資源分配由spark集羣的Master負責
    • yarn
      • 把spark程序提交到yarn中運行，整個任務的資源分配由yarn中的ResourceManager負責
    • mesos
      • apache開源的一個相似於yarn的資源調度平臺
• Master
  • Spark集羣的主節點，負責任務資源的分配
• Worker
  • Spark集羣的從節點，負責任務計算的節點
• Executor
  • 是一個在worker節點啓動的進程
• Task
  • spark任務的以task線程的方式運行在worker節點的executor進程中的

3. RDD是什麼

• RDD （Resilient Distributed Dataset）叫作彈性分佈式數據集，是Spark中最基本的抽象單位。它表明的是是一個不可變的、可分區的、裏面元素能夠並行計算的數據集合。
  • Resilient 彈性的，表示RDD中的數據既能夠保存在磁盤上也能保存在內存中
  • Distibuted 分佈式的，表示RDD的數據是分佈式存儲的，方便後期的各類計算
  • Dataset 一個數據集合，能夠存儲不少數據

4. RDD的五大屬性

• A list of partitions
  • 一個分區列表，數據集的基本組成單位
    • 這裏表示的是一個RDD可能會有多個分區，每一個分區會存儲該RDD的一部分數據，Spark中任務是以task線程的方式運行的，一個分區就對應一個task
• A function of computing each split
  • 一個用來計算每一個分區的函數
    • Spark中RDD的計算是以分區爲單位的

• A list of dependencies on other RDDs

  • 一個RDD會依賴於其餘多個RDD

    這裏是說RDD和RDD之間是有依賴關係的，spark任務的容錯機制就是根據這個特性（血統）而來

• Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)

  • 一個Patitioner，即RDD的分區函數(可選項)

    spark中實現了兩種類型的分區函數
        1 基於哈希的HashPartitioner，（key.hashcode % 分區數 = 分區號）
        2 基於範圍的RangePartitioner
    
        只有對於key-value的RDD，而且產生shuffle，纔會有Partitioner
        非key-value的RDD的Partitioner的值是None

• Optionally, a list of preferred locations to compute each split in (e.g. block locations for an HDFS file)

  • 一個列表，存儲每一個Partition的優先位置（可選項）

    spark任務在調度的時候會優先考慮存有數據的節點開啓計算任務，以減小數據的網絡傳輸，提成計算效率

5. RDD算子分類

• transformation（轉換）
  • 根據已經存在的RDD轉換生成另一個新的RDD，它是延遲加載，不會當即執行
  • 如
    • map、flatMap、reduceByKey
• action（動做）
  • 會觸發任務的運行
    • 將RDD計算的結果數據返回給Driver端，或者保存到外部存儲介質（磁盤、內存、HDFS）
  • 如
    • collect、saveAsTextFile

6. RDD常見的算子操做

6.1 transformation算子

轉換	含義
map(func)	返回一個新的RDD，該RDD由每個輸入元素通過func函數轉換後組成
filter(func)	返回一個新的RDD，該函數由通過func函數計算而且返回值爲true的輸入元素組成
flatMap(func)	相似於map，但每個輸入元素能夠被映射爲0或多個輸出元素（因此func應該返回一個序列，而不是單一的元素）
mapPartitions(func)	相似於map，但獨立地在RDD的每個分片上運行，所以在類型爲T的RDD上運行時，func的函數類型必須是Iterator[T] => Iterator[U]
mapPartitionsWithIndex(func)	相似於mapPartitions，但func帶有一個整數參數表示分片的索引值，所以在類型爲T的RDD上運行時，func的函數類型必須是Iterator[T] => Iterator[U]
union(otherDataset)	對源RDD和參數RDD求並集，並返回一個新的RDD
intersection(otherDataset)	對源RDD和參數RDD求交集，並返回一個新的RDD
distinct([numTasks]))	對源RDD進行去重以後返回一個新的RDD
groupByKey([numTasks])	在一個(k,v)類型的RDD上調用，返回一個(k,v)的RDD
reduceByKey(func, [numTasks])	在一個(k,v)類型的RDD上調用，返回一個(k,v)的RDD，使用指定的reduce函數，將相同key的值聚合到一塊兒，與groupByKey相似，reduce任務的個數能夠經過第二個參數來設置
sortByKey([ascending], [numTasks])	在一個(k,v)的RDD上調用，k必須實現Ordered接口，返回一個按照key進行排序的(k,v)RDD
sortBy(func,[ascending], [numTasks])	與sortByKey相似，可是更靈活，能夠自定義排序func
join(otherDataset, [numTasks])	在類型爲(k,v)和(k,w)的RDD上調用，返回一個相同 key對應的全部元素對在一塊兒的(k,(v,w))的RDD
cogroup(otherDataset, [numTasks])	在類型爲(K,V)和(K,W)的RDD上調用，返回一個(K,(Iterable ,Iterable ))類型的RDD
coalesce(numPartitions)	減小RDD的分區數到指定值
repartition(numPartitions)	從新給RDD分區
repartitionAndSortWithinPartitions(partitioner)	從新給RDD分區，而且每一個分區內以記錄的key排序

6.2 action算子

動做	含義
reduce(func)	reduce將RDD中元素前兩個傳給輸入函數，產生一個新的return值，新產生的return值與RDD中下一個元素（第三個元素）組成兩個元素，再被傳給輸入函數，直到最後只有一個值爲止。
collect()	在驅動程序中，以數組的形式返回數據集的全部元素
count()	返回RDD的元素個數
first()	返回RDD的第一個元素（相似於take(1)）
take(n)	返回一個由數據集的前n個元素組成的數組
takeOrdered(n, [ordering])	返回天然順序或者自定義順序的前 n 個元素
saveAsTextFile(path)	將數據集中的元素以textFile的形式保存到HDFS文件系統或者其餘支持的文件系統，對於每一個元素，Spark將會調用toString方法，將它裝換爲文件中的文本
saveAsSequenceFile(path)	將數據集中的元素以Hadoop sequenceFile的格式保存到指定的目錄，能夠是HDFS或者其餘Hadoop支持的文件系統
saveAsObjectFile(path)	將數據集的元素以Java序列化的方式保存到指定的目錄下
countByKey()	針對(k,v)類型的RDD，返回一個(k,int)的map，表示每個key對應的元素個數
foreach(func)	在數據集上每一個元素上，運行函數func
foreachPartition(func)	在數據集的每一個分區上，運行函數func