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

• √Spark的关注度分析
• √Spark的特点
• √Spark的一些重要概念
• √Spark组件概览
• √Spark运行架构概览
• √Spark编程小试牛刀

Spark的关注热度分析

概况

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

近1年内全球关于Spark、Hadoop及Flink的搜索趋势&＃xff0c;如下&＃xff1a;

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

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

分析

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

Spark的特点

• 速度快Apache Spark使用DAG调度程序、查询优化器和物理执行引擎&＃xff0c;为批处理和流处理提供了高性能。
• 易于使用支持使用Java&＃xff0c;Scala&＃xff0c;Python&＃xff0c;R和SQL快速编写应用程序。Spark提供了80多个高级操作算子&＃xff0c;可轻松构建并行应用程序。
• 通用性Spark提供了非常丰富的生态栈&＃xff0c;包括SQL查询、流式计算、机器学习和图计算等组件&＃xff0c;这些组件可以无缝整合在一个应用中&＃xff0c;通过一站部署&＃xff0c;可以应对多种复杂的计算场景
• 运行模式多样Spark可以使用Standalone模式运行&＃xff0c;也可以运行在Hadoop&＃xff0c;Apache Mesos&＃xff0c;Kubernetes等环境中运行。并且可以访问HDFS、Alluxio、Apache Cassandra、Apache HBase、Apache Hive等多种数据源中的数据。

Spark的一些重要概念

• RDD弹性分布式数据集(Resilient Distributed Dataset)&＃xff0c;是分布式内存的一个抽象概念&＃xff0c;提供了一种高度受限的共享内存模型
• DAG有向无环图(Directed Acyclic Graph),反映RDD之间的依赖关系
• Application用户编写的Spark程序&＃xff0c;由 driver program 和 executors 组成
• Application jar用户编写的应用程序JAR包
• Driver program用程序main()函数的进程&＃xff0c;可以创建SparkContext
• Cluster manager集群管理器&＃xff0c;属于一个外部服务&＃xff0c;用于资源请求分配(如&＃xff1a;standalone manager, Mesos, YARN)
• Deploy mode部署模式&＃xff0c;决定Driver进程在哪里运行。如果是cluster模式&＃xff0c;会由框架本身在集群内部某台机器上启动Driver进程。如果是client模式&＃xff0c;会在提交程序的机器上启动Driver进程
• Worker node集群中运行应用程序的节点Executor运行在Worknode节点上的一个进程&＃xff0c;负责运行具体的任务&＃xff0c;并为应用程序存储数据
• Task运行在executor中的工作单元
• Job一个job包含多个RDD及一些列的运行在RDD之上的算子操作&＃xff0c;job需要通过action操作进行触发(比如save、collect等)
• Stage每一个作业会被分成由一些列task组成的stage&＃xff0c;stage之间会相互依赖

Spark组件概览

Spark生态系统主要包括Spark Core、SparkSQL、SparkStreaming、MLlib和GraphX等组件&＃xff0c;具体如下图所示&＃xff1a;

• Spark CoreSpark core是Spark的核心&＃xff0c;包含了Spark的基本功能&＃xff0c;如内存计算、任务调度、部署模式、存储管理等。SparkCore提供了基于RDD的API是其他高级API的基础&＃xff0c;主要功能是实现批处理。
• Spark SQLSpark SQL主要是为了处理结构化和半结构化数据而设计的&＃xff0c;SparkSQL允许用户在Spark程序中使用SQL、DataFrame和DataSetAPI查询结构化数据&＃xff0c;支持Java、Scala、Python和R语言。由于DataFrame API提供了统一的访问各种数据源的方式(包括Hive、Avro、Parquet、ORC和JDBC)&＃xff0c;用户可以通过相同的方式连接任何数据源。另外&＃xff0c;Spark SQL可以使用hive的元数据&＃xff0c;从而实现了与Hive的完美集成&＃xff0c;用户可以将Hive的作业直接运行在Spark上。Spark SQL可以通过spark-sql的shell命令访问。
• SparkStreamingSparkStreaming是Spark很重要的一个模块&＃xff0c;可实现实时数据流的可伸缩&＃xff0c;高吞吐量&＃xff0c;容错流处理。在内部&＃xff0c;其工作方式是将实时输入的数据流拆分为一系列的micro batch&＃xff0c;然后由Spark引擎进行处理。SparkStreaming支持多种数据源&＃xff0c;如kafka、Flume和TCP套接字等
• MLlibMLlib是Spark提供的一个机器学习库&＃xff0c;用户可以使用Spark API构建一个机器学习应用&＃xff0c;Spark尤其擅长迭代计算&＃xff0c;性能是Hadoop的100倍。该lib包含了常见机器学习算法&＃xff0c;比如逻辑回归、支持向量机、分类、聚类、回归、随机森林、协同过滤、主成分分析等。
• GraphXGraphX是Spark中用于图计算的API&＃xff0c;可认为是Pregel在Spark上的重写及优化&＃xff0c;GraphX性能良好&＃xff0c;拥有丰富的功能和运算符&＃xff0c;能在海量数据上自如地运行复杂的图算法。GraphX内置了许多图算法&＃xff0c;比如著名的PageRank算法。

Spark运行架构概览

从整体来看&＃xff0c;Spark应用架构包括以下几个主要部分&＃xff1a;

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

在Standalone模式下&＃xff0c;运行架构如下图所示&＃xff1a;

Driver program

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

Cluster Manager

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

Worker node

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

Executor

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

Task

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

SparkContext

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

Spark编程小试牛刀

Spark实现分组取topN案例

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

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

问题&＃xff1a;使用sparkcore&＃xff0c;求每个订单中成交额最大的商品id

实现代码

import org.apache.spark.sql.Rowimport org.apache.spark.sql.hive.HiveContextimport org.apache.spark.sql.types.{StringType, StructField, StructType}import org.apache.spark.{SparkConf, SparkContext}object TopOrderItemCluster {  def main(args: Array[String]): Unit &＃61; {    val conf &＃61; new SparkConf().setAppName("top n order and item")    val sc &＃61; new SparkContext(conf)    val hctx &＃61; new HiveContext(sc)    val orderData &＃61; sc.textFile("data.txt")    val splitOrderData &＃61; orderData.map(_.split(","))    val mapOrderData &＃61; splitOrderData.map { arrValue &＃61;>      val orderID &＃61; arrValue(0)      val itemID &＃61; arrValue(1)      val total &＃61; arrValue(2).toDouble      (orderID, (itemID, total))    }    val groupOrderData &＃61; mapOrderData.groupByKey()    /**      ***groupOrderData.foreach(x &＃61;> 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 &＃61; groupOrderData.map(tupleData &＃61;> {      val orderid &＃61; tupleData._1      val maxTotal &＃61; tupleData._2.toArray.sortWith(_._2 > _._2).take(1)      (orderid, maxTotal)    }    )    topOrderData.foreach(value &＃61;>      println("最大成交额的订单ID为&＃xff1a;" &＃43; value._1 &＃43; " ,对应的商品ID为&＃xff1a;" &＃43; value._2(0)._1)      /**        ***最大成交额的订单ID为&＃xff1a;Order_00003 ,对应的商品ID为&＃xff1a;Pdt_01        ***最大成交额的订单ID为&＃xff1a;Order_00002 ,对应的商品ID为&＃xff1a;Pdt_05        ***最大成交额的订单ID为&＃xff1a;Order_00001 ,对应的商品ID为&＃xff1a;Pdt_01        */          )    //构造出元数据为Row的RDD    val RowOrderData &＃61; topOrderData.map(value &＃61;> Row(value._1, value._2(0)._1))    //构建元数据    val structType &＃61; StructType(Array(      StructField("orderid", StringType, false),      StructField("itemid", StringType, false))    )    //转换成DataFrame    val orderDataDF &＃61; 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 &＃39;&＃39;")      hctx.sql("INSERT INTO orderid_itemid SELECT * FROM tmptable")  }}

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

总结

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