Spark簡介

簡介:

  spark是一個內存計算框架 , 解決了mapreduce在迭代與交互方面的不足

  迭代: 多輪算法計算形式,pagerank,邏輯迴歸等

  交互: 實時數據,數據倉庫查詢等

spark重要概念:

  RDD: 

  彈性分佈式數據集, 彈性(數據多節點分片)分佈式(集羣節點)數據集(默認加載至內存)

  操做:

    轉換(transformation) rdd->newrdd

    # 注意此類操做不會提交做業

    map(func) 造成新的RDD集

    filter(func) 過濾結果成爲新的RDD集

    flatmap(func) 映射爲多於0的輸出元素

    groupbykey(key) 以Key爲組的數據集

    reducebykey(func) 以Key爲組的數據集在通過func的處理

    join(newrdd) 鏈接兩個數據集 k,v k,w => k,(v,w)

    sortbykey(desc) 排序數據集

    動做(action) 結果處理

    # 注意此類操做將提交做業

    reduce(func) func函數輸入兩個返回一個結果 

    collect 返回操做後的數據元素

    count 元素個數統計

    countByKey 按Key統計每組的數目

    saveastextfile 以文件形式保存計算結果集

    saveassequencefile 以序列文件保存計算結果集

  操做分類:

  value:

  1->1:

  map, flatmap, mappartitions, glom

  n->1:

  union類型一致合併, cartesian分區笛卡爾積

  n->n:

  groupby

  n屬於m:

  filter, distinct, subtract, sample, takesample

  cache:

  cache僅內存, persist可決定緩存位置

  key-value:

  1->1:

  mapValues

  union:

  combineByKey, reduceByKey, partitionBy分區策略, cogroup協同劃分

  join:

  join(cogroup,flatMapValues), leftOutJoin, rightOutJoin

  action:

  無輸出:

  foreach

  HDFS:

  saveAsTextFile, saveAsObjectFile

  Scala:

  collect, collectAsMap, reduceByKeyLocally, count, lookup, top(top,take,takeOrdered,first), reduce, fold, aggregate

  DataType:

spark編程概念:

共享變量: 

廣播變量(boarkcast不可修改) .value

累加器(accumulator可累加)  .value

初始化: 

構建SparkContext(集羣地址,做業名稱,sparkhome,做業依賴項)

RDD類型: 

並行集合與hadoop數據集

並行集合: scala集合,可分片slice(parallelize)

hadoop集合: 文本,序列(K,V),inputformat

(以後都可作map/reduce操做)

  Cache: 

  RDD轉換後可緩存persist,cache,可容錯

  cache級別: 僅內存,內存磁盤,內存序列化kryo,內存磁盤序列化,僅磁盤,

spark基本架構:

應用程序間環境隔離 , 資源管理(YARN , Mesos) + 應用程序管理 . 程序運行過程當中對RDD分片的優化(窄fusion , 寬全局barrier)

spark運行流程:

 提交應用 -> DAG依賴流程化 -> 執行DAG至完成 -> 銷燬運行環境

提交應用:

 構建運行環境: 粗(一次YARN,Mesos).細(屢次Mesos)

 啓動RDD管理器 BlockManager

DAG依賴流程化:

 分解寬窄依賴: 歸組

fusion優化,不須要物化

執行DAG至完成:

 數據本地化: 基架感知

 推測執行: 另起資源比較

銷燬運行環境:

回收資源

spark on YARN:

客戶端 提交任務 至 ApplicationMaster 申請資源 向 Nodemanager

SAM 內部有 ClusterScheduler(cpu核數) 有了資源 啓動 StandaloneExecutorBackend(container大小) 註冊 akka 等待任務 以後執行 完畢回收資源

Mesos介紹:

集羣資源管理器 twitter,douban

客戶端scheduler -> MesosSchedulerDriver -> MesosMaster(zk) -> MesosSlave(Executor)

master: 核心, 資源接入,分配Allocator,其餘框架的管理

salve: 接收命令,報告狀態(任務,資源)

framework scheduler: MesosSchedulerDriver第三方實現接入mesos, 雙層調度(資源給框架,框架自定任務)  --->  相似hadoop jobtrack

framework executor: MesosExecutorDriver第三方執行實現  ---> 相似hadoop tasktrack

第三方框架註冊流程:

第三方實現MesosSchedulerDriver,啓動SchedulerProcess服務,MasterDetector對象(探測master通知schedulerprocess),sp會向master發送註冊消息,sp確認,framework確認

mesos資源分配: 主資源公平調度算法,多維度資源,受權接入框架

設計: 資源拒絕(slave), 資源過濾(salve), 資源回收(master)

mesos容錯設計:

master: zk

salve: 心跳,快照運行目錄

mesos vs YARN:

設計目標一致:　通用集羣資源管理方案

容錯一致: zk,多主

在線升級: YARN的salve不支持

調度模型一致: 雙層

調度算法: YARN除了DRF還支持容量與失敗

資源隔離一致: cgroups

支持框架一致: mapreduce,strorm,spark等

資源分配顆粒度: YARN不支持動態資源分配

網絡庫:

rpc            serialization

  libprocess     protocal buffer

  avro           avro

  Thrift         idl

  socket kryo

  libprocess設計: 消息(類型) -> 消息隊列 -> 消息處理分發器(dispatcher) -> 消息處理    高效,易用

spark工做機制:

應用運行提交:

spark任務調度分配:

spark io: 序列化(進程actor,NIO,netty,磁盤kyro)

壓縮(snappy,LZF)

spark net:RPC, RMI, WEB Service, EJB

AKKA

spark 容錯: 血統, checkpoint 

spark編程:

下載源碼包 , 安裝maven , ./make-distribution.sh --skip-java-test構建本地模式

安裝sbt打包工具

編寫代碼:

python: 

from pyspark import SparkContext

scala: 

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

編寫sbt依賴文件:

name := "Simple Project"

version := "1.0"

scalaVersion := "2.10"

libraryDependencies += "org.apache.spark" %% "spark-core" % "1.1.0"

sbt打包: sbt package

提交應用: /bin/spark-submit --class "SimpleApp" --master local test/target/scala-2.10/simple-project_2.10-1.0.jar