Apache Spark最初在2009年诞生于美国加州大学伯克利分校的APM实验室,并于2010年开源,如今是Apache软件基金会下的顶级开源项目之一。Spark的目标是设计一种编程模型,能够快速地进行数据分析。Spark提供了内存计算,减少了IO开销。另外Spark是基于Scala编写的,提供了交互式的编程体验。经过10年的发展,Spark成为了炙手可热的大数据处理平台,目前最新的版本是Spark3.0。本文主要是对Spark进行一个总体概览式的介绍,后续内容会对具体的细节进行展开讨论。本文的主要内容包括:

  • Spark的关注度分析
  • Spark的特点
  • Spark的一些重要概念
  • Spark组件概览
  • Spark运行架构概览
  • Spark编程初体验

Spark的关注热度分析

概况

下图展示了近1年内在国内关于Spark、Hadoop及Flink的搜索趋势

近1年内全球关于Spark、Hadoop及Flink的搜索趋势,如下:

近1年国内关于Spark、Hadoop及Flink的搜索热度区域分布情况(按Flink搜索热度降序排列):

近1年全球关于Spark、Hadoop及Flink的搜索热度区域分布情况(按Flink搜索热度降序排列):

分析

从上面的4幅图可以看出,近一年无论是在国内还是全球,关于Spark的搜索热度始终是比Hadoop和Flink要高。近年来Flink发展迅猛,其在国内有阿里的背书,Flink天然的流处理特点使其成为了开发流式应用的首选框架。可以看出,虽然Flink在国内很火,但是放眼全球,热度仍然不及Spark。所以学习并掌握Spark技术仍然是一个不错的选择,技术有很多的相似性,如果你已经掌握了Spark,再去学习Flink的话,相信你会有种似曾相识的感觉。

Spark的特点

  • 速度快

    Apache Spark使用DAG调度程序、查询优化器和物理执行引擎,为批处理和流处理提供了高性能。

  • 易于使用

    支持使用Java,Scala,Python,R和SQL快速编写应用程序。Spark提供了80多个高级操作算子,可轻松构建并行应用程序。

  • 通用性

    Spark提供了非常丰富的生态栈,包括SQL查询、流式计算、机器学习和图计算等组件,这些组件可以无缝整合在一个应用中,通过一站部署,可以应对多种复杂的计算场景

  • 运行模式多样

    Spark可以使用Standalone模式运行,也可以运行在Hadoop,Apache Mesos,Kubernetes等环境中运行。并且可以访问HDFS、Alluxio、Apache Cassandra、Apache HBase、Apache Hive等多种数据源中的数据。

Spark的一些重要概念

  • RDD

    弹性分布式数据集(Resilient Distributed Dataset),是分布式内存的一个抽象概念,提供了一种高度受限的共享内存模型

  • DAG

    有向无环图(Directed Acyclic Graph),反映RDD之间的依赖关系

  • Application

    用户编写的Spark程序,由 driver program 和 executors 组成

  • Application jar
    用户编写的应用程序JAR包

  • Driver program
    用程序main()函数的进程,可以创建SparkContext

  • Cluster manager
    集群管理器,属于一个外部服务,用于资源请求分配(如:standalone manager, Mesos, YARN)

  • Deploy mode

    部署模式,决定Driver进程在哪里运行。如果是cluster模式,会由框架本身在集群内部某台机器上启动Driver进程。如果是client模式,会在提交程序的机器上启动Driver进程

  • Worker node

    集群中运行应用程序的节点Executor运行在Worknode节点上的一个进程,负责运行具体的任务,并为应用程序存储数据

  • Task
    运行在executor中的工作单元

  • Job
    一个job包含多个RDD及一些列的运行在RDD之上的算子操作,job需要通过action操作进行触发(比如save、collect等)

  • Stage
    每一个作业会被分成由一些列task组成的stage,stage之间会相互依赖

Spark组件概览

Spark生态系统主要包括Spark Core、SparkSQL、SparkStreaming、MLlib和GraphX等组件,具体如下图所示:

  • Spark Core

    Spark core是Spark的核心,包含了Spark的基本功能,如内存计算、任务调度、部署模式、存储管理等。SparkCore提供了基于RDD的API是其他高级API的基础,主要功能是实现批处理。

  • Spark SQL

    Spark SQL主要是为了处理结构化和半结构化数据而设计的,SparkSQL允许用户在Spark程序中使用SQL、DataFrame和DataSetAPI查询结构化数据,支持Java、Scala、Python和R语言。由于DataFrame API提供了统一的访问各种数据源的方式(包括Hive、Avro、Parquet、ORC和JDBC),用户可以通过相同的方式连接任何数据源。另外,Spark SQL可以使用hive的元数据,从而实现了与Hive的完美集成,用户可以将Hive的作业直接运行在Spark上。Spark SQL可以通过spark-sql的shell命令访问。

  • SparkStreaming

    SparkStreaming是Spark很重要的一个模块,可实现实时数据流的可伸缩,高吞吐量,容错流处理。在内部,其工作方式是将实时输入的数据流拆分为一系列的micro batch,然后由Spark引擎进行处理。SparkStreaming支持多种数据源,如kafka、Flume和TCP套接字等

  • MLlib

    MLlib是Spark提供的一个机器学习库,用户可以使用Spark API构建一个机器学习应用,Spark尤其擅长迭代计算,性能是Hadoop的100倍。该lib包含了常见机器学习算法,比如逻辑回归、支持向量机、分类、聚类、回归、随机森林、协同过滤、主成分分析等。

  • GraphX

    GraphX是Spark中用于图计算的API,可认为是Pregel在Spark上的重写及优化,GraphX性能良好,拥有丰富的功能和运算符,能在海量数据上自如地运行复杂的图算法。GraphX内置了许多图算法,比如著名的PageRank算法。

Spark运行架构概览

从整体来看,Spark应用架构包括以下几个主要部分:

  • Driver program
  • Master node
  • Work node
  • Executor
  • Tasks
  • SparkContext

Standalone模式下,运行架构如下图所示:

Driver program

Driver program是Spark应用程序的main()函数(创建SparkContext和Spark会话)。运行Driver进程的节点称之为Driver node,Driver进程与集群管理器(Cluster Manager)进行通信,向Executor发送调度的task。

Cluster Manager

称之为集群管理器,主要用于管理集群。常见的集群管理器包括YARN、Mesos和Standalone,Standalone集群管理器包括两个长期运行的后台进程,其中一个是在Master节点,另外一个是在Work节点。在后续集群部署模式篇,将详细探讨这一部分的内容,此处先有有一个大致印象即可。

Worker node

熟悉Hadoop的朋友应该知道,Hadoop包括namenode和datanode节点。Spark也类似,Spark将运行具体任务的节点称之为Worker node。该节点会向Master节点汇报当前节点的可用资源,通常在每一台Worker node上启动一个work后台进程,用于启动和监控Executor。

Executor

Master节点分配资源,使用集群中的Work node创建Executor,Driver使用这些Executor分配运行具体的Task。每一个应用程序都有自己的Executor进程,使用多个线程执行具体的Task。Executor主要负责运行任务和保存数据。

Task

Task是发送到Executor中的工作单元

SparkContext

SparkContext是Spark会话的入口,用于连接Spark集群。在提交应用程序之前,首先需要初始化SparkContext,SparkContext隐含了网络通信、存储体系、计算引擎、WebUI等内容。值得注意的是,一个JVM进程中只能有一个SparkContext,如果想创建新的SparkContext,需要在原来的SparkContext上调用stop()方法。

Spark编程小试牛刀

Spark实现分组取topN案例

描述:在HDFS上有订单数据order.txt文件,文件字段的分割符号”,”,其中字段依次表示订单id,商品id,交易额。样本数据如下:

Order_00001,Pdt_01,222.8
Order_00001,Pdt_05,25.8
Order_00002,Pdt_03,522.8
Order_00002,Pdt_04,122.4
Order_00002,Pdt_05,722.4
Order_00003,Pdt_01,222.8

问题:使用sparkcore,求每个订单中成交额最大的商品id

实现代码

import org.apache.spark.sql.Row
import org.apache.spark.sql.hive.HiveContext
import org.apache.spark.sql.types.{StringType, StructField, StructType}
import org.apache.spark.{SparkConf, SparkContext}

object TopOrderItemCluster {
def main(args: Array[String]): Unit = {
val conf = new SparkConf().setAppName("top n order and item")
val sc = new SparkContext(conf)
val hctx = new HiveContext(sc)
val orderData = sc.textFile("data.txt")
val splitOrderData = orderData.map(_.split(","))
val mapOrderData = splitOrderData.map { arrValue =>
val orderID = arrValue(0)
val itemID = arrValue(1)
val total = arrValue(2).toDouble
(orderID, (itemID, total))
}
val groupOrderData = mapOrderData.groupByKey()

/**
***groupOrderData.foreach(x => println(x))
***(Order_00003,CompactBuffer((Pdt_01,222.8)))
***(Order_00002,CompactBuffer((Pdt_03,522.8), (Pdt_04,122.4), (Pdt_05,722.4)))
***(Order_00001,CompactBuffer((Pdt_01,222.8), (Pdt_05,25.8)))
*/

val topOrderData = groupOrderData.map(tupleData => {
val orderid = tupleData._1
val maxTotal = tupleData._2.toArray.sortWith(_._2 > _._2).take(1)
(orderid, maxTotal)
}
)
topOrderData.foreach(value =>
println("最大成交额的订单ID为:" + value._1 + " ,对应的商品ID为:" + value._2(0)._1)

/**
***最大成交额的订单ID为:Order_00003 ,对应的商品ID为:Pdt_01
***最大成交额的订单ID为:Order_00002 ,对应的商品ID为:Pdt_05
***最大成交额的订单ID为:Order_00001 ,对应的商品ID为:Pdt_01
*/

)
//构造出元数据为Row的RDD
val RowOrderData = topOrderData.map(value => Row(value._1, value._2(0)._1))
//构建元数据
val structType = StructType(Array(
StructField("orderid", StringType, false),
StructField("itemid", StringType, false))
)
//转换成DataFrame
val orderDataDF = hctx.createDataFrame(RowOrderData, structType)
// 将数据写入Hive
orderDataDF.registerTempTable("tmptable")
hctx.sql("CREATE TABLE IF NOT EXISTS orderid_itemid(orderid STRING,itemid STRING) ROW FORMAT DELIMITED FIELDS TERMINATED BY '\\t'")
hctx.sql("INSERT INTO orderid_itemid SELECT * FROM tmptable")
}

}

将上述代码打包,提交到集群运行,可以进入hive cli或者spark-sql的shell查看Hive中的数据。

总结

本文主要从整体上对Spark进行了介绍,主要包括Spark的搜索热度分析、Spark的主要特点、Spark的一些重要概念以及Spark的运行架构,最后给出了一个Spark编程案例。本文是Spark系列分享的第一篇,可以先感受一下Spark的全局面貌,下一篇将分享Spark Core编程指南。