|
【编者按】时迄今日，Spark已作为大数据行业最火的一个开源项目，具备高性能、易于运用等特性。然而做为一个青年的开源项目，其运用上存在的挑战也不可为不大，这儿为大众分享SciSpike软件架构师Ashwini Kuntamukkala在Dzone上进行的Spark入门总结（虽然有些地区基于的是Spark 1.0版本，但仍然值得阅读）——Apache Spark：An Engine for Large-Scale Data Processing，由OneAPM工程师翻译。
本文聚焦Apache Spark入门，认识其在大数据行业的地位，覆盖Apache Spark的安装及应用程序的创立，并解释有些平常的行径和操作。
1、 为何要运用Apache Spark
时下，咱们正处在一个“大数据”的时代，每时每刻，都有各样类型的数据被生产。而这里紫外，数据增幅的速度亦在明显增多。从广义上看，这些数据包括交易数据、社交媒身体容（例如文本、图像和视频）以及传感器数据。那样，为何要在这些内容上投入如此多精力，其原由无非便是从海量数据中提取洞见能够对生活和生产实践进行很好的指点。
在几年前，仅有少部分机构持有足够的技术力量和资金去储存和挖掘海量数据，并对其挖掘从而得到洞见。然而，被雅虎2009年开源的Apache Hadoop对这一情况产生了颠覆性的冲击——经过运用商用服务器构成的集群大幅度地降低了海量数据处理的门槛。因此呢，许多行业（例如Health care、Infrastructure、Finance、Insurance、Telematics、Consumer、Retail、Marketing、E-commerce、Media、 Manufacturing和Entertainment）起始了Hadoop的征程，走上了海量数据提取价值的道路。着眼Hadoop，其重点供给了两个方面的功能： 经过水平扩展商用主机，HDFS供给了一个低价的方式对海量数据进行容错存储。MapReduce计算范例，供给了一个简单的编程模型来挖掘数据并得到洞见。下图展示了MapReduce的数据处理流程，其中一个Map-Reduce step的输出将做为下一个典型Hadoop job的输入结果。 
在全部过程中，中间结果会借助磁盘传递，因此呢对比计算，海量的Map-Reduced作业都受限于IO。然而针对ETL、数据整合和清理这般的用例来讲，IO约束并不会产生很大的影响，由于这些场景对数据处理时间常常不会有较高的需要。然而，在现实世界中，一样存在许多对延时需求较为苛刻的用例，例如： 对流数据进行处理来做近实时分析。举个例子，经过分析点击流数据做视频举荐，从而加强用户的参与度。在这个用例中，研发者必须在精度和延时之间做平衡。在大型数据集上进行交互式分析，数据专家能够在数据集上做ad-hoc查找。
毫无疑问，历经数年发展，Hadoop生态圈中的丰富工具已深受用户喜欢，然而这儿仍然存在众多问题给运用带来了挑战：
1.每一个用例都需要多个区别的技术堆栈来支撑，在区别运用场景下，海量的处理方法常常捉襟见肘。
2.在生产环境中公司常常需要精通数门技术。
3.许多技术存在版本兼容性问题。
4.没法在并行job中更快地共享数据。
而经过Apache Spark，以上问题迎刃而解！Apache Spark是一个轻量级的内存集群计算平台，经过区别的组件来支撑批、流和交互式用例，如下图。 
2、 关于Apache Spark
Apache Spark是个开源和兼容Hadoop的集群计算平台。由加州大学伯克利分校的AMPLabs研发，做为Berkeley Data Analytics Stack（BDAS）的一部分，当下由大数据机构Databricks保驾护航，更加是Apache旗下的顶级项目，下图表示了Apache Spark堆栈中的区别组件。 
Apache Spark的5大优良：
1.更高的性能，由于数据被加载到集群主机的分布式内存中。数据可以被快速的转换迭代，并缓存用以后续的频繁拜访需要。非常多对Spark感兴趣的伴侣可能亦会听过这般一句话——在数据所有加载到内存的状况下，Spark能够比Hadoop快100倍，在内存不足存放所有数据的状况下快Hadoop 10倍。 
2.经过创立在Java、Scala、Python、SQL（应对交互式查找）的标准API以方便各行各业运用，同期还含有海量开箱即用的设备学习库。
3.与现有Hadoop v1 (SIMR) 和2.x (YARN) 生态兼容，因此呢公司能够进行无缝迁移。 
4.方便下载和安装。方便的shell（REPL: Read-Eval-Print-Loop）能够对API进行交互式的学习。
5.借助高等级的架构加强生产力，从而能够讲精力放到计算上。
同期，Apache Spark由Scala实现，代码非常简洁。
3、安装Apache Spark
下表列出了有些重要链接和先决要求：
如图6所示，Apache Spark的安排方式包含standalone、Hadoop V1 SIMR、Hadoop 2 YARN/Mesos。Apache Spark需要必定的Java、Scala或Python知识。这儿，咱们将专注standalone配置下的安装和运行。
1.安装JDK 1.6+、Scala 2.10+、Python [2.6,3] 和sbt
2.下载Apache Spark 1.0.1 Release
3.在指定目录下Untar和Unzip spark-1.0.1.tgz akuntamukkala@localhost~/Downloads$ pwd
/Users/akuntamukkala/Downloads akuntamukkala@localhost~/Downloads$ tar -zxvf spark- 1.0.1.tgz -C /Users/akuntamukkala/spark4.运行sbt创立Apache Spark akuntamukkala@localhost~/spark/spark-1.0.1$ pwd /Users/akuntamukkala/spark/spark-1.0.1 akuntamukkala@localhost~/spark/spark-1.0.1$ sbt/sbt assembly5.发布Scala的Apache Spark standalone REPL /Users/akuntamukkala/spark/spark-1.0.1/bin/spark-shell倘若是Python /Users/akuntamukkala/spark/spark-1.0.1/bin/ pyspark4、Apache Spark的工作模式
Spark引擎供给了在集群中所有主机上进行分布式内存数据处理的能力，下图表示了一个典型Spark job的处理流程。 
下图表示了Apache Spark怎样在集群中执行一个作业。 
Master掌控数据怎样被分割，利用了数据本地性，并在Slaves上跟踪所有分布式计算。在某个Slave不可用时，其存储的数据会分配给其他可用的Slaves。虽然当下（1.0.1版本）Master还存在单点故障，但后期必然会被修复。
5、弹性分布式数据集（Resilient Distributed Dataset，RDD）
弹性分布式数据集（RDD，从Spark 1.3版本起始已被DataFrame替代）是Apache Spark的核心理念。它是由于数据构成的不可变分布式集合，其重点进行两个操作：transformation和action。Transformation是类似在RDD上做 filter、map或union 以生成另一个RDD的操作，而action则是count、first、take(n)、collect 等促发一个计算并返回值到Master或稳定存储系统的操作。Transformations通常都是lazy的，直到action执行后才会被执行。Spark Master/Driver会保留RDD上的Transformations。这般一来，倘若某个RDD丢失（亦便是salves宕掉），它能够快速和方便地转换到集群中存活的主机上。这亦便是RDD的弹性所在。
下图展示了Transformation的lazy： 
咱们能够经过下面示例来理解这个概念：从文本中发掘5个最常用的word。下图表示了一个可能的处理方法。
在上面命令中，咱们对文本进行读取并且创立字符串的RDD。每一个条目表率了文本中的1行。 scala> val hamlet = sc.textFile(“/Users/akuntamukkala/temp/gutenburg.txt”)
hamlet: org.apache.spark.rdd.RDD[String] = MappedRDD[1] at textFile at <console>:12scala> val topWordCount = hamlet.flatMap(str=>str.split(“ “)). filter(!_.isEmpty).map(word=>(word,1)).reduceByKey(_+_).map{case (word, count) => (count, word)}.sortByKey(false)
topWordCount: org.apache.spark.rdd.RDD[(Int, String)] = MapPartitionsRDD[10] at sortByKey at <console>:141. 经过以上命令咱们能够发掘这个操作非常简单——经过简单的Scala API来连接transformations和actions。
2. 可能存在某些words被1个以上空格分隔的状况，引起有些words是空字符串，因此呢需要运用filter(!_.isEmpty)将它们过滤掉。
3.每一个word都被映射成一个键值对：map(word=>(word,1))。
4.为了合计所有计数，这儿需要调用一个reduce过程——reduceByKey(_+_)。 _+_ 能够非常方便地为每一个key赋值。
5. 咱们得到了words以及各自的counts，下一步需要做的是按照counts排序。在Apache Spark，用户只能按照key排序，而不是值。因此呢，这儿需要运用map{case (word, count) => (count, word)}将(word, count)流转到(count, word)。
6. 需要计算最常用的5个words，因此呢需要运用sortByKey(false)做一个计数的递减排序。
以上命令包括了一个.take(5) (an action operation, which triggers computation)和在 /Users/akuntamukkala/temp/gutenburg.txt文本中输出10个最常用的words。在Python shell中用户能够实现一样的功能。
RDD lineage能够经过toDebugString（一个值得记住的操作）来跟踪。 scala> topWordCount.take(5).foreach(x=>println(x))
(1044,the)
(730,and)
(679,of)
(648,to)
(511,I)常用的Transformations：
Transformation & PurposeExample & Resultfilter(func)Purpose:new RDD by selecting those data elements on which func returns truescala> val rdd = sc.parallelize(List(“ABC”,”BCD”,”DEF”)) scala> val filtered = rdd.filter(_.contains(“C”)) scala> filtered.collectResult:Array[String] = Array(ABC, BCD)map(func) Purpose:return new RDD by applying func on each data elementscala> val rdd=sc.parallelize(List(1,2,3,4,5)) scala> val times2 = rdd.map(_*2) scala> times2.collectResult:Array[Int] = Array(2, 4, 6, 8, 10)flatMap(func)Purpose:Similar to map but func returns a Seq instead of a value. For example, mapping a sentence into a Seq of wordsscala> val rdd=sc.parallelize(List(“Spark is awesome”,”It is fun”)) scala> val fm=rdd.flatMap(str=>str.split(“ “)) scala> fm.collectResult:Array[String] = Array(Spark, is, awesome, It, is, fun)reduceByKey(func,[numTasks])Purpose:To aggregate values of a key using a function. “numTasks” is an optional parameter to specify number of reduce tasksscala> val word1=fm.map(word=>(word,1)) scala> val wrdCnt=word1.reduceByKey(_+_) scala> wrdCnt.collectResult:Array[(String, Int)] = Array((is,2), (It,1), (awesome,1), (Spark,1), (fun,1))groupByKey([numTasks]) Purpose:To convert (K,V) to (K,Iterable<V>)scala> val cntWrd = wrdCnt.map{case (word, count) => (count, word)} scala> cntWrd.groupByKey.collectResult:Array[(Int, Iterable[String])] = Array((1,ArrayBuffer(It, awesome, Spark, fun)), (2,ArrayBuffer(is)))distinct([numTasks])Purpose:Eliminate duplicates from RDDscala> fm.distinct.collect Result:Array[String] = Array(is, It, awesome, Spark, fun)
常用的集合操作：
Transformation and PurposeExample and Resultunion Purpose:
new RDD containing all elements from source RDD and argument.Scala> val rdd1=sc.parallelize(List(‘A’,’B’))
scala> val rdd2=sc.parallelize(List(‘B’,’C’))
scala> rdd1.union(rdd2).collect Result:
Array[Char] = Array(A, B, B, C)intersection Purpose:
new RDD containing only common elements from source RDD and argument.Scala> rdd1.intersection(rdd2).collect Result:
Array[Char] = Array(B)cartesian Purpose:
new RDD cross product of all elements from source RDD and argumentScala> rdd1.cartesian(rdd2).collect Result:
Array[(Char, Char)] = Array((A,B), (A,C), (B,B), (B,C))subtract Purpose:new RDD created by removing data elements in source RDD in common with argumentscala> rdd1.subtract(rdd2).collect Result:
Array[Char] = Array(A)join(RDD,[numTasks]) Purpose:
When invoked on (K,V) and (K,W), this operation creates a new RDD of (K, (V,W))scala> val personFruit = sc.parallelize(Seq((“Andy”, “Apple”), (“Bob”, “Banana”), (“Charlie”, “Cherry”), (“Andy”,”Apricot”)))
scala> val personSE = sc.parallelize(Seq((“Andy”, “Google”), (“Bob”, “Bing”), (“Charlie”, “Yahoo”), (“Bob”,”AltaVista”)))
scala> personFruit.join(personSE).collect Result:
Array[(String, (String, String))] = Array((Andy,(Apple,Google)), (Andy,(Apricot,Google)), (Charlie,(Cherry,Yahoo)), (Bob,(Banana,Bing)), (Bob,(Banana,AltaVista)))cogroup(RDD,[numTasks]) Purpose:
To convert (K,V) to (K,Iterable)scala> personFruit.cogroup(personSe).collect
Result:Array[(String, (Iterable[String], Iterable[String]))] = Array((Andy,(ArrayBuffer(Apple, Apricot),ArrayBuffer(google))), (Charlie,(ArrayBuffer(Cherry),ArrayBuffer(Yahoo))), (Bob,(ArrayBuffer(Banana),ArrayBuffer(Bing, AltaVista))))
更加多transformations信息，请查看 http://spark.apache.org/docs/latest/programming-guide.html#transformations
常用的actions
Action & PurposeExample & Resultcount Purpose:get the number of data elements in the RDDscala> val rdd = sc.parallelize(list(‘A’,’B’,’c’)) scala> rdd.countResult:long = 3collect Purpose:get all the data elements in an RDD as an arrayscala> val rdd = sc.parallelize(list(‘A’,’B’,’c’)) scala> rdd.collectResult:Array[char] = Array(A, B, c)reduce(func) Purpose:Aggregate the data elements in an RDD using this function which takes two arguments and returns onescala> val rdd = sc.parallelize(list(1,2,3,4)) scala> rdd.reduce(_+_)Result:Int = 10take (n) Purpose:: fetch first n data elements in an RDD. computed by driver program.Scala> val rdd = sc.parallelize(list(1,2,3,4)) scala> rdd.take(2)Result:Array[Int] = Array(1, 2)foreach(func)Purpose:execute function for each data element in RDD. usually used to update an accumulator(discussed later) or interacting with external systems.Scala> val rdd = sc.parallelize(list(1,2,3,4)) scala> rdd.foreach(x=>println(“%s*10=%s”. format(x,x*10)))Result:1*10=10 4*10=40 3*10=30 2*10=20firstPurpose:retrieves the first data element in RDD. Similar to take(1)scala> val rdd = sc.parallelize(list(1,2,3,4)) scala> rdd.first Result:Int = 1saveAsTextFile(path) Purpose:Writes the content of RDD to a text file or a set of text files to local file system/ HDFSscala> val hamlet = sc.textFile(“/users/akuntamukkala/ temp/gutenburg.txt”) scala> hamlet.filter(_.contains(“Shakespeare”)). saveAsTextFile(“/users/akuntamukkala/temp/ filtered”)Result:akuntamukkala@localhost~/temp/filtered$ ls _SUCCESS part-00000 part-00001
更加多actions参见 http://spark.apache.org/docs/latest/programming-guide.html#actions
6、RDD持久性
Apache Spark中一个重点的能力便是在集群内存中持久化/缓存RDD。这将明显地提高交互速度。下表表示了Spark中各样选项。
Storage LevelPurposeMEMORY_ONLY (Default level)This option stores RDD in available cluster memory as deserialized Java objects. Some partitions may not be cached if there is not enough cluster memory. Those partitions will be recalculated on the fly as needed.MEMORY_AND_DISKThis option stores RDD as deserialized Java objects. If RDD does not fit in cluster memory, then store those partitions on the disk and read them as needed.MEMORY_ONLY_SERThis options stores RDD as serialized Java objects (One byte array per partition). This is more CPU intensive but saves memory as it is more space efficient. Some partitions may not be cached. Those will be recalculated on the fly as needed.MEMORY_ONLY_DISK_SERThis option is same as above except that disk is used when memory is not sufficient.DISC_ONLYThis option stores the RDD only on the diskMEMORY_ONLY_2, MEMORY_AND_DISK_2, etc.Same as other levels but partitions are replicated on 2 slave nodes
上面的存储等级能够经过RDD. cache操作上的 persist操作拜访，能够方便地指定MEMORY_ONLY选项。关于持久化等级的更加多信息，能够拜访这儿 http://spark.apache.org/docs/latest/programming-guide.html#rdd-persistence。
Spark运用Least Recently Used (LRU)算法来移除缓存中旧的、不常用的RDD，从而释放出更加多可用内存。一样还供给了一个unpersist 操作来强制移除缓存/持久化的RDD。
7、变量共享
Accumulators。Spark供给了一个非常方便地途径来避免可变的计数器和计数器同步问题——Accumulators。Accumulators在一个Spark context中经过默认值初始化，这些计数器在Slaves节点上可用，然则Slaves节点不可对其进行读取。它们的功效便是来获取原子更新，并将其转发到Master。Master是独一能够读取和计算所有更新合集的节点。举个例子： akuntamukkala@localhost~/temp$ cat output.log
error
warning
info
trace
error
info
info
scala> val nErrors=sc.accumulator(0.0)
scala> val logs = sc.textFile(“/Users/akuntamukkala/temp/output.log”)
scala> logs.filter(_.contains(“error”)).foreach(x=>nErrors+=1)
scala> nErrors.value
Result:Int = 2Broadcast Variables。实质生产中，经过指定key在RDDs上对数据进行合并的场景非常平常。在这种状况下，很可能会显现给slave nodes发送大体积数据集的状况，让其负责托管需要做join的数据。因此呢，这儿很可能存在巨大的性能瓶颈，由于网络IO比内存拜访速度慢100倍。为认识决这个问题，Spark供给了Broadcast Variables，如其名叫作同样，它会向slave nodes进行广播。因此呢，节点上的RDD操作能够快速拜访Broadcast Variables值。举个例子，期望计算一个文件中所有路线项的运输成本。经过一个look-up table指定每种运输类型的成本，这个look-up table就能够做为Broadcast Variables。 akuntamukkala@localhost~/temp$ cat packagesToShip.txt ground
express
media
priority
priority
ground
express
media
scala> val map = sc.parallelize(Seq((“ground”,1),(“med”,2), (“priority”,5),(“express”,10))).collect.toMap
map: scala.collection.immutable.Map[String,Int] = Map(ground -> 1, media -> 2, priority -> 5, express -> 10)
scala> val bcMailRates = sc.broadcast(map)以上命令中，咱们创立了一个broadcast variable，基于服务类别成本的map。 scala> val pts = sc.textFile(“/Users/akuntamukkala/temp/packagesToShip.txt”)在以上命令中，咱们经过broadcast variable的mailing rates来计算运输成本。 scala> pts.map(shipType=>(shipType,1)).reduceByKey(_+_). map{case (shipType,nPackages)=>(shipType,nPackages*bcMailRates. value(shipType))}.collect经过以上命令，咱们运用accumulator来累加所有运输的成本。仔细信息可经过下面的PDF查看 http://ampcamp.berkeley.edu/wp-content/uploads/2012/06/matei-zaharia-amp-camp-2012-advanced-spark.pdf。
8、Spark SQL
经过Spark Engine，Spark SQL供给了一个方便的途径来进行交互式分析，运用一个被叫作为SchemaRDD类型的RDD。SchemaRDD能够经过已有RDDs创立，或者其他外边数据格式，例如Parquet files、JSON数据，或在Hive上运行HQL。SchemaRDD非常类似于RDBMS中的表格。一旦数据被导入SchemaRDD，Spark引擎就能够对它进行批或流处理。Spark SQL供给了两种类型的Contexts——SQLContext和HiveContext，扩展了SparkContext的功能。
SparkContext供给了到简单SQL parser的拜访，而HiveContext则供给了到HiveQL parser的拜访。HiveContext准许企业利用已有的Hive基本设备。
这儿看一个简单的SQLContext示例。
下面文本中的用户数据经过“|”来分割。 John Smith|38|M|201 East Heading Way #2203,Irving, TX,75063 Liana Dole|22|F|1023 West Feeder Rd, Plano,TX,75093 Craig Wolf|34|M|75942 Border Trail,Fort Worth,TX,75108 John Ledger|28|M|203 Galaxy Way,Paris, TX,75461 Joe Graham|40|M|5023 Silicon Rd,London,TX,76854定义Scala case class来暗示每一行： case class Customer(name:String,age:Int,gender:String,address: String)下面的代码片段表现了怎样运用SparkContext来创立SQLContext，读取输入文件，将每一行都转换成SparkContext中的一条记录，并经过简单的SQL语句来查找30岁以下的男性用户。 val sparkConf = new SparkConf.setAppName(“Customers”)
val sc = new SparkContext(sparkConf)
val sqlContext = new SQLContext(sc)
val r = sc.textFile(“/Users/akuntamukkala/temp/customers.txt”) val records = r.map(_.split(‘|’))
val c = records.map(r=>Customer(r(0),r(1).trim.toInt,r(2),r(3))) c.registerAsTable(“customers”)sqlContext.sql(“select * from customers where gender=’M’ and age <
30”).collect.foreach(println) Result:[John Ledger,28,M,203 Galaxy Way,Paris,
TX,75461]更加多运用SQL和HiveQL的示例请拜访下面链接
https://spark.apache.org/docs/latest/sql-programming-guide.html、 https://databricks-training.s3.amazonaws.com/data-exploration-using-spark-sql.html。
9、Spark Streaming
Spark Streaming供给了一个可扩展、容错、有效的途径来处理流数据，同期还利用了Spark的简易编程模型。从真正道理上讲，Spark Streaming会将流数据转换成micro batches，从而将Spark批处理编程模型应用到流用例中。这种统一的编程模型让Spark能够很好地整合批量处理和交互式流分析。下图表示了Spark Streaming能够从区别数据源中读取数据进行分析。 
Spark Streaming中的核心抽象是Discretized Stream（DStream）。DStream由一组RDD构成，每一个RDD都包括了规按时间（可配置）流入的数据。图12很好地展示了Spark Streaming怎样经过将流入数据转换成一系列的RDDs，再转换成DStream。每一个RDD都包括两秒（设定的区间长度）的数据。在Spark Streaming中，最小长度能够设置为0.5秒，因此呢处理延时能够达到1秒以下。
Spark Streaming一样供给了 window operators，它有助于更有效率在一组RDD（ a rolling window of time）上进行计算。同期，DStream还供给了一个API，其操作符（transformations和output operators）能够帮忙用户直接操作RDD。下面不妨看向包括在Spark Streaming下载中的一个简单示例。示例是在Twitter流中找出趋势hashtags，详见下面代码。 spark-1.0.1/examples/src/main/scala/org/apache/spark/examples/streaming/TwitterPopularTags.scala
val sparkConf = new SparkConf.setAppName(“TwitterPopularTags”)
val ssc = new StreamingContext(sparkConf, Seconds(2))
val stream = TwitterUtils.createStream(ssc, None, filters)以上代码用于创立Spark Streaming Context。Spark Streaming将在DStream中创立一个RDD，包括了每2秒流入的tweets。 val hashTags = stream.flatMap(status => status.getText.split(“ “).filter(_.startsWith(“#”)))上述代码片段将Tweet转换成一组words，并过滤出所有以a#开头的。 val topCounts60 = hashTags.map((_, 1)).reduceByKeyAndWindow(_ + _, Seconds(60)).map{case (topic, count) => (count, topic)}. transform(_.sortByKey(false))以上代码展示了怎样整合计算60秒内一个hashtag流入的总次数。 topCounts60.foreachRDD(rdd => {
val topList = rdd.take(10)
println(“\nPopular topics in last 60 seconds (%s
total):”.format(rdd.count)) topList.foreach{case (count, tag) => println(“%s (%s
tweets)”.format(tag, count))} })上面代码将找出top 10趋势tweets，而后将其打印。 ssc.start以上代码让Spark Streaming Context 起始检索tweets。一块聚焦有些常用操作，假设咱们正在从一个socket中读入流文本。 al lines = ssc.socketTextStream(“localhost”, 9999, StorageLevel.MEMORY_AND_DISK_SER)  
更加多operators请拜访 http://spark.apache.org/docs/latest/streaming-programming-guide.html#transformations
Spark Streaming持有海量强大的output operators，例如上文说到的 foreachRDD，认识更加多可拜访 http://spark.apache.org/docs/latest/streaming-programming-guide.html#output-operations。
十、附加学习资源
|
发表于 2024-10-10 14:34:42