公司网站招聘模板,如何夸奖一个网站做的好,深圳苏州企业网站建设服务公司,1个空间做2个网站Spark 主要用于替代Hadoop中的 MapReduce 计算模型。存储依然可以使用 HDFS#xff0c;但是中间结果可以存放在内存中#xff1b;调度可以使用 Spark 内置的#xff0c;也可以使用更成熟的调度系统 YARN 等。
Spark有完善的生态圈#xff1a; Spark Core#xff1a;实现了…Spark 主要用于替代Hadoop中的 MapReduce 计算模型。存储依然可以使用 HDFS但是中间结果可以存放在内存中调度可以使用 Spark 内置的也可以使用更成熟的调度系统 YARN 等。
Spark有完善的生态圈 Spark Core实现了 Spark 的基本功能包含 RDD、任务调度、内存管理、错误恢复、与存储系统交互等模块。Spark SQLSpark 用来操作结构化数据的程序包。通过 Spark SQL我们可以使用 SQL 操作数据。Spark StreamingSpark 提供的对实时数据进行流式计算的组件。提供了用来操作数据流的 API。Spark MLlib提供常见的机器学习(ML)功能的程序库。包括分类、回归、聚类、协同过滤等还提供了模型评估、数据导入等额外的支持功能。GraphX(图计算)Spark 中用于图计算的 API性能良好拥有丰富的功能和运算符能在海量数据上自如地运行复杂的图算法。集群管理器Spark 设计为可以高效地在一个计算节点到数千个计算节点之间伸缩计算。Structured Streaming处理结构化流,统一了离线和实时的 API。 Spark 特点 快
与 Hadoop 的 MapReduce 相比Spark 基于内存的运算要快 100 倍以上基于硬盘的运算也要快 10 倍以上。Spark 通过基于内存来高效处理数据流。
易用
Spark 支持 Java、Python、R 和 Scala 的 API还支持超过 80 种高级算法使用户可以快速构建不同的应用。而且 Spark 支持交互式的 Python 和 Scala 的 shell可以非常方便地在这些 shell 中使用 Spark 集群来验证解决问题的方法。
通用
Spark 提供了统一的解决方案。Spark 可以用于批处理、交互式查询(Spark SQL)、实时流处理(Spark Streaming)、机器学习(Spark MLlib)和图计算(GraphX)这些不同类型的处理都可以在同一个应用中无缝使用。
兼容性
Spark 可以非常方便地与其他的开源产品进行融合。比如Spark 可以使用 Hadoop 的 YARN 和 Apache Mesos 作为它的资源管理和调度器并且可以处理所有 Hadoop 支持的数据包括 HDFS、HBase 和 Cassandra 等。这对于已经部署 Hadoop 集群的用户特别重要因为不需要做任何数据迁移就可以使用 Spark 的强大处理能力。 Spark Core RDD
MapReduce 框架采用非循环式的数据流模型会把中间结果写入到 HDFS 中带来了大量的数据复制、磁盘 IO 和序列化开销。且这些框架只能支持一些特定的计算模式(map/reduce)并没有提供一种通用的数据抽象。因此出现了RDD这个概念。
RDD(Resilient Distributed Dataset)叫做弹性分布式数据集是 Spark 中最基本的数据抽象代表一个不可变、可分区、里面的元素可并行计算的集合。
RDD 不实际存储真正要计算的数据而是记录了数据的位置在哪里数据的转换关系(调用了什么方法传入什么函数)。
RDD单词拆解
Resilient 它是弹性的RDD 里面的中的数据可以保存在内存中或者磁盘里面RDD的数据默认存放在内存中但是当内存资源不足时spark会自动将RDD数据写入磁盘。比如某结点内存只能处理20W数据那么这20W数据就会放入内存中计算剩下10W放到磁盘中。RDD的弹性体现在于RDD上自动进行内存和磁盘之间权衡和切换的机制Distributed 它里面的元素是分布式存储的可以用于分布式计算Dataset: 它是一个集合可以存放很多元素。
RDD 属性 RDD 的源码描述如下
其含义如下
A list of partitions 一组分片(Partition)/一个分区(Partition)列表即数据集的基本组成单位。对于 RDD 来说每个分片都会被一个计算任务处理分片数决定并行度。用户可以在创建 RDD 时指定 RDD 的分片个数如果没有指定那么就会采用默认值。A function for computing each split 一个函数会被作用在每一个分区。Spark 中 RDD 的计算是以分片为单位的compute 函数会被作用到每个分区上。A list of dependencies on other RDDs 一个 RDD 会依赖于其他多个 RDD。RDD 的每次转换都会生成一个新的 RDD所以 RDD 之间就会形成类似于流水线一样的前后依赖关系。在部分分区数据丢失时Spark 可以通过这个依赖关系重新计算丢失的分区数据而不是对 RDD 的所有分区进行重新计算。(Spark 的容错机制)Optionally, a Partitioner for key-value RDDs (e.g. to say that the RDD is hash-partitioned)可选项对于 KV 类型的 RDD 会有一个 Partitioner即 RDD 的分区函数默认为 HashPartitioner。Optionally, a list of preferred locations to compute each split on (e.g. block locations for an HDFS file)可选项,一个列表存储存取每个 Partition 的优先位置(preferred location)。对于一个 HDFS 文件来说这个列表保存的就是每个 Partition 所在的块的位置。按照移动数据不如移动计算的理念Spark 在进行任务调度的时候会尽可能选择那些存有数据的 worker 节点来进行任务计算。
总结
RDD 是一个数据集的表示不仅表示了数据集还表示了这个数据集从哪来如何计算主要属性包括分区列表、计算函数、依赖关系、分区函数(默认是 hash)、最佳位置。分区列表、分区函数、最佳位置这三个属性其实说的就是数据集在哪在哪计算更合适如何分区计算函数、依赖关系这两个属性其实说的是数据集怎么来的。
RDD API
RDD 的创建方式 ① 由外部存储系统的数据集创建包括本地的文件系统还有所有 Hadoop 支持的数据集比如 HDFS、Cassandra、HBase 等
val rdd1 sc.textFile(hdfs://node1:8020/wordcount/input/words.txt)
② 通过已有的 RDD 经过算子转换生成新的 RDD
val rdd2rdd1.flatMap(_.split( ))
③ 由一个已经存在的 Scala 集合创建
val rdd3 sc.parallelize(Array(1,2,3,4,5,6,7,8))
或者
val rdd4 sc.makeRDD(List(1,2,3,4,5,6,7,8))
makeRDD 方法底层调用了 parallelize 方法
RDD 算子
RDD 的算子分为两类:
Transformation转换操作:返回一个新的 RDDAction动作操作:返回值不是 RDD(无返回值或返回其他的)
Transformation转换算子 Action 动作算子 统计操作 Transformation和action算子有什么区别
Transformation 变换/转换这种变换并不触发提交作业完成作业中间过程处理。Transformation 操作是延迟计算的也就是说从一个RDD 转换生成另一个 RDD 的转换操作不是马上执行需要等到有 Action 操作的时候才会真正触发运算
Action 行动算子这类算子会触发 SparkContext 提交 Job 作业。 RDD 持久化/缓存
某些 RDD 的计算或转换可能会比较耗费时间如果这些 RDD 后续还会频繁的被使用到那么可以将这些 RDD 进行持久化/缓存
val rdd1 sc.textFile(hdfs://node01:8020/words.txt)
val rdd2 rdd1.flatMap(xx.split( )).map((_,1)).reduceByKey(__)
rdd2.cache //缓存/持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读取HDFS的文件,rdd2会真正执行持久化
rdd2.sortBy(_._2,false).collect//触发action,会去读缓存中的数据,执行速度会比之前快,因为rdd2已经持久化到内存中了
RDD 通过 persist 或 cache 方法可以将前面的计算结果缓存但是并不是这两个方法被调用时立即缓存而是触发后面的 action 时该 RDD 将会被缓存在计算节点的内存中并供后面重用。通过查看 RDD 的源码发现 cache 最终也是调用了 persist 无参方法(默认存储只存在内存中)。
存储级别 默认的存储级别都是仅在内存存储一份Spark 的存储级别还有好多种存储级别在 object StorageLevel 中定义的。
总结
RDD 持久化/缓存的目的是为了提高后续操作的速度缓存的级别有很多默认只存在内存中,开发中使用 memory_and_disk只有执行 action 操作的时候才会真正将 RDD 数据进行持久化/缓存实际开发中如果某一个 RDD 后续会被频繁的使用可以将该 RDD 进行持久化/缓存这里当小数据量的时候缓存能提升效率但数据大的时候内存放不下就会报溢出
RDD 容错机制Checkpoint
持久化/缓存可以把数据放在内存中虽然是快速的但是也是最不可靠的也可以把数据放在磁盘上也不是完全可靠的例如磁盘会损坏等。
怎么解决
Checkpoint 的产生就是为了更加可靠的数据持久化在Checkpoint的时候一般把数据放在在 HDFS 上这就天然的借助了 HDFS 天生的高容错、高可靠来实现数据最大程度上的安全实现了 RDD 的容错和高可用。用法如下
SparkContext.setCheckpointDir(目录) //HDFS的目录RDD.checkpoint
总结
开发中如何保证数据的安全性性及读取效率可以对频繁使用且重要的数据先做缓存/持久化再做 checkpint 操作。
持久化和 Checkpoint 的区别
位置Persist 和 Cache 只能保存在本地的磁盘和内存中(或者堆外内存–实验中) Checkpoint 可以保存数据到 HDFS 这类可靠的存储上。生命周期Cache 和 Persist 的 RDD 会在程序结束后会被清除或者手动调用 unpersist 方法 Checkpoint 的 RDD 在程序结束后依然存在不会被删除。
RDD 的依赖关系
RDD有两种依赖分别为宽依赖(wide dependency/shuffle dependency)和窄依赖(narrow dependency) :
从上图可以看到 窄依赖父 RDD 的一个分区只会被子 RDD 的一个分区依赖宽依赖父 RDD 的一个分区会被子 RDD 的多个分区依赖(涉及到 shuffle) 对于窄依赖窄依赖的多个分区可以并行计算窄依赖的一个分区的数据如果丢失只需要重新计算对应的分区的数据就可以了。
对于宽依赖划分 Stage(阶段)的依据:对于宽依赖,必须等到上一阶段计算完成才能计算下一阶段 DAG 的生成和划分 Stage
DAG
DAG(Directed Acyclic Graph 有向无环图)指的是数据转换执行的过程有方向无闭环(其实就是 RDD 执行的流程)
原始的 RDD 通过一系列的转换操作就形成了 DAG 有向无环图任务执行时可以按照 DAG 的描述执行真正的计算(数据被操作的一个过程)。
DAG 的边界:
开始通过 SparkContext 创建的 RDD结束触发 Action一旦触发 Action 就形成了一个完整的 DAG。
DAG 划分Stage 从上图可以看出
一个 Spark 程序可以有多个 DAG(有几个 Action就有几个 DAG上图最后只有一个 Action图中未表现,那么就是一个 DAG);一个 DAG 可以有多个 Stage(根据宽依赖/shuffle 进行划分)同一个 Stage 可以有多个 Task 并行执行(task 数分区数如上图Stage1 中有三个分区 P1、P2、P3对应的也有三个 Task)可以看到这个 DAG 中只 reduceByKey 操作是一个宽依赖Spark 内核会以此为边界将其前后划分成不同的 Stage在图中 Stage1 中从 textFile 到 flatMap 到 map 都是窄依赖这几步操作可以形成一个流水线操作通过 flatMap 操作生成的 partition 可以不用等待整个 RDD 计算结束而是继续进行 map 操作这样大大提高了计算的效率。
所以总结下stage是如何划分的
从hdfs中读取文件后创建 RDD 对象DAGScheduler模块介入运算计算RDD之间的依赖关系。RDD之间的依赖关系就形成了DAG
每一个JOB被分为多个Stage划分Stage的一个主要依据是当前计算因子的输入是否是确定的如果是则将其分在同一个Stage避免多个Stage之间的消息传递开销。 因此spark划分stage的整体思路是从后往前推遇到宽依赖就断开划分为一个stage遇到窄依赖就将这个RDD加入该stage中
为什么要划分 Stage?
为了并行计算。
一个复杂的业务逻辑如果有 shuffle那么就意味着前面阶段产生结果后才能执行下一个阶段即下一个阶段的计算要依赖上一个阶段的数据。那么我们按照 shuffle 进行划分(也就是按照宽依赖就行划分)就可以将一个 DAG 划分成多个 Stage/阶段在同一个 Stage 中会有多个算子操作可以形成一个 pipeline 流水线流水线内的多个平行的分区可以并行执行。
总结
Spark 会根据 shuffle/宽依赖使用回溯算法来对 DAG 进行 Stage 划分从后往前遇到宽依赖就断开遇到窄依赖就把当前的 RDD 加入到当前的 stage/阶段中。
RDD累加器和广播变量
在默认情况下当 Spark 在集群的多个不同节点的多个任务上并行运行一个函数时它会把函数中涉及到的每个变量在每个任务上都生成一个副本。但是有时候需要在多个任务之间共享变量或者在任务(Task)和任务控制节点(Driver Program)之间共享变量。
为了满足这种需求Spark 提供了两种类型的变量
累加器 accumulators累加器支持在所有不同节点之间进行累加计算(比如计数或者求和)。广播变量 broadcast variables广播变量用来把变量在所有节点的内存之间进行共享在每个机器上缓存一个只读的变量而不是为机器上的每个任务都生成一个副本。
累加器通常在向 Spark 传递函数时比如使用 map() 函数或者用filter()传条件时可以使用驱动器程序中定义的变量但是集群中运行的每个任务都会得到这些变量的一份新的副本更新这些副本的值也不会影响驱动器中的对应变量。这时使用累加器就可以实现我们想要的效果:
语法
val xx: Accumulator[Int] sc.accumulator(0)
示例代码
import org.apache.spark.rdd.RDD
import org.apache.spark.{Accumulator, SparkConf, SparkContext}object AccumulatorTest {def main(args: Array[String]): Unit {val conf: SparkConf new SparkConf().setAppName(wc).setMaster(local[*])val sc: SparkContext new SparkContext(conf)sc.setLogLevel(WARN)//使用scala集合完成累加var counter1: Int 0;var data Seq(1,2,3)data.foreach(x counter1 x )println(counter1)//6println()//使用RDD进行累加var counter2: Int 0;val dataRDD: RDD[Int] sc.parallelize(data) //分布式集合的[1,2,3]dataRDD.foreach(x counter2 x)println(counter2)//0//注意上面的RDD操作运行结果是0//因为foreach中的函数是传递给Worker中的Executor执行,用到了counter2变量//而counter2变量在Driver端定义的,在传递给Executor的时候,各个Executor都有了一份counter2//最后各个Executor将各自个x加到自己的counter2上面了,和Driver端的counter2没有关系//那这个问题得解决啊!不能因为使用了Spark连累加都做不了了啊!//如果解决?---使用累加器val counter3: Accumulator[Int] sc.accumulator(0)dataRDD.foreach(x counter3 x)println(counter3)//6}
}
广播变量 关键词sc.broadcast()
import org.apache.spark.broadcast.Broadcast
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}object BroadcastVariablesTest {def main(args: Array[String]): Unit {val conf: SparkConf new SparkConf().setAppName(wc).setMaster(local[*])val sc: SparkContext new SparkContext(conf)sc.setLogLevel(WARN)//不使用广播变量val kvFruit: RDD[(Int, String)] sc.parallelize(List((1,apple),(2,orange),(3,banana),(4,grape)))val fruitMap: collection.Map[Int, String] kvFruit.collectAsMap//scala.collection.Map[Int,String] Map(2 - orange, 4 - grape, 1 - apple, 3 - banana)val fruitIds: RDD[Int] sc.parallelize(List(2,4,1,3))//根据水果编号取水果名称val fruitNames: RDD[String] fruitIds.map(xfruitMap(x))fruitNames.foreach(println)//注意:以上代码看似一点问题没有,但是考虑到数据量如果较大,且Task数较多,//那么会导致,被各个Task共用到的fruitMap会被多次传输//应该要减少fruitMap的传输,一台机器上一个,被该台机器中的Task共用即可//如何做到?---使用广播变量//注意:广播变量的值不能被修改,如需修改可以将数据存到外部数据源,如MySQL、Redisprintln()val BroadcastFruitMap: Broadcast[collection.Map[Int, String]] sc.broadcast(fruitMap)val fruitNames2: RDD[String] fruitIds.map(xBroadcastFruitMap.value(x))fruitNames2.foreach(println)}
} Spark SQL Hive 是将 SQL 转为 MapReduce。
SparkSQL 可以理解成是将 SQL 解析成“RDD 优化” 再执行。
在学习Spark SQL前需要了解数据分类。
数据分为如下几类 总结
RDD 主要用于处理非结构化数据 、半结构化数据、结构化SparkSQL 主要用于处理结构化数据(较为规范的半结构化数据也可以处理)
Spark SQL 数据抽象
DataFrame 和 DataSet
Spark SQL数据抽象可以分为两类
① DataFrame
DataFrame 是一种以 RDD 为基础的分布式数据集类似于传统数据库的二维表格带有 Schema 元信息(可以理解为数据库的列名和类型)。DataFrame RDD 泛型 SQL 的操作 优化
② DataSet
DataSet是DataFrame的进一步发展它比RDD保存了更多的描述信息概念上等同于关系型数据库中的二维表它保存了类型信息是强类型的提供了编译时类型检查。调用 Dataset 的方法先会生成逻辑计划然后被 spark 的优化器进行优化最终生成物理计划然后提交到集群中运行DataFrame Dateset[Row]
RDD、DataFrame、DataSet的关系如下 RDD[Person]以 Person 为类型参数但不了解其内部结构。DataFrame提供了详细的结构信息 schema 列的名称和类型。这样看起来就像一张表了。DataSet[Person]不光有 schema 信息还有类型信息。
举例 假设 RDD 中的两行数据长这样
RDD[Person] 那么 DataFrame 中的数据长这样 因为DataFrame RDD[Person] 泛型 Schema SQL 操作 优化
那么 Dataset 中的数据长这样 因为Dataset[Person] DataFrame 泛型
Dataset 也可能长这样:Dataset[Row] 即 DataFrame DataSet[Row]
总结 DataFrame RDD - 泛型 Schema SQL 优化DataSet DataFrame 泛型DataSet RDD Schema SQL 优化 Spark SQL 应用
创建 DataFrame/DataSet
方式一读取本地文件
① 在本地创建一个文件有 id、name、age 三列用空格分隔然后上传到 hdfs 上。
vim /root/person.txt 内容如下 1 zhangsan 20 2 lisi 29 3 wangwu 25 4 zhaoliu 30 5 tianqi 35 6 kobe 40 ② 打开 spark-shell
spark/bin/spark-shell
##创建 RDDval lineRDD sc.textFile(hdfs://node1:8020/person.txt).map(_.split( )) //RDD[Array[String]]
③ 定义 case class(相当于表的 schema)
case class Person(id:Int, name:String, age:Int)
④ 将 RDD 和 case class 关联
val personRDD lineRDD.map(x Person(x(0).toInt, x(1), x(2).toInt)) //RDD[Person]
⑤ 将 RDD 转换成 DataFrame
val personDF personRDD.toDF //DataFrame
也可以通过 SparkSession 构建 DataFrame
val dataFramespark.read.text(hdfs://node1:8020/person.txt)
dataFrame.show //注意直接读取的文本文件没有完整schema信息
dataFrame.printSchema
两种查询风格DSL 和 SQL DSL风格示例
personDF.select(personDF.col(name)).show
personDF.select(personDF(name)).show
personDF.select(col(name)).show
personDF.select(name).show
SQL 风格示例:
spark.sql(select * from t_person).show
总结
DataFrame 和 DataSet 都可以通过RDD来进行创建不管是 DataFrame 还是 DataSet 都可以注册成表之后可以使用 SQL 进行查询也可以使用 DSL Spark Streaming Spark Streaming 是一个基于 Spark Core 之上的实时计算框架可以从很多数据源消费数据并对数据进行实时的处理具有高吞吐量和容错能力强等特点。 Spark streaming内部的基本工作原理是接受实时输入数据流然后将数据拆分成batch比如每收集一秒的数据封装成一个batch然后将每个batch交给spark的计算引擎进行处理最后会生产处一个结果数据流其中的数据也是一个一个的batch组成的。
Spark Streaming 的特点
易用可以像编写离线批处理一样去编写流式程序支持 java/scala/python 语言。容错SparkStreaming 在没有额外代码和配置的情况下可以恢复丢失的工作。易整合到 Spark 体系流式处理与批处理和交互式查询相结合。
整体流程
① Spark Streaming 中会有一个接收器组件 Receiver作为一个长期运行的 task 跑在一个 Executor 上Receiver 接收外部的数据流形成 input DStream。② DStream 会被按照时间间隔划分成一批一批的 RDD当批处理间隔缩短到秒级时便可以用于处理实时数据流时间间隔的大小可以由参数指定一般设在 500 毫秒到几秒之间。③ 对 DStream 进行操作就是对 RDD 进行操作计算处理的结果可以传给外部系统。④ 接受到实时数据后给数据分批次然后传给 Spark Engine 处理最后生成该批次的结果。
数据抽象
Spark Streaming 的基础抽象是 DStream(Discretized Stream离散化数据流连续不断的数据流)代表持续性的数据流和经过各种 Spark 算子操作后的结果数据流。
可以从以下多个角度深入理解 DStream
① DStream 本质上就是一系列时间上连续的 RDD每个RDD包含了一个时间段的数据 ② 对 DStream 的数据的进行操作也是按照 RDD 为单位来进行的 ③ 容错性底层 RDD 之间存在依赖关系DStream 直接也有依赖关系RDD 具有容错性那么 DStream 也具有容错性。
④ 准实时性/近实时性
Spark Streaming 将流式计算分解成多个 Spark Job对于每一时间段数据的处理都会经过 Spark DAG 图分解以及 Spark 的任务集的调度过程。对于目前版本的 Spark Streaming 而言其最小的 Batch Size 的选取在 0.5~5 秒钟之间。
所以 Spark Streaming 能够满足流式准实时计算场景对实时性要求非常高的如高频实时交易场景则不太适合。
总结 简单来说 DStream 就是对 RDD 的封装你对 DStream 进行操作就是对 RDD 进行操作。对于 DataFrame/DataSet/DStream 来说本质上都可以理解成 RDD。 Spark 两种核心 Shuffle 在 MapReduce 框架中Shuffle 阶段是连接 Map 与 Reduce 之间的桥梁 Map 阶段通过 Shuffle 过程将数据输出到 Reduce 阶段中。由于 Shuffle 涉及磁盘的读写和网络 I/O因此 Shuffle 性能的高低直接影响整个程序的性能。Spark 也有 Map 阶段和 Reduce 阶段因此也会出现Shuffle。
哪些spark算子会有shuffle
去重distinct排序groupByKeyreduceByKey等重分区repartitioncoalesce集合或者表操作interectionjoi
Spark Shuffle 分为两种
一种是基于 Hash 的 Shuffle另一种是基于 Sort 的 Shuffle。
在spark-1.6版本之前一直使用HashShuffle在spark-1.6版本之后使用Sort-Base Shuffle因为HashShuffle存在的不足所以就替换了HashShuffle.
Hash Shuffle
HashShuffleManager shuffle write 阶段主要就是在一个 stage 结束计算之后为了下一个 stage 可以执行 shuffle 类的算子比如 reduceByKey而将每个 task 处理的数据按 key 进行“划分”。所谓“划分”就是对相同的 key 执行 hash 算法从而将相同 key 都写入同一个磁盘文件中而每一个磁盘文件都只属于下游 stage 的一个 task。在将数据写入磁盘之前会先将数据写入内存缓冲中当内存缓冲填满之后才会溢写到磁盘文件中去。
下一个 stage 的 task 有多少个当前 stage 的每个 task 就要创建多少份磁盘文件。比如: 下一个 stage 总共有 100 个 task那么当前 stage 的每个 task 都要创建 100 份磁盘文件。如果当前 stage 有 50 个 task总共有 10 个 Executor每个 Executor 执行 5 个 task那么每个 Executor 上总共就要创建 500 个磁盘文件所有 Executor 上会创建 5000 个磁盘文件。 由此可见未经优化的 shuffle write 操作所产生的磁盘文件的数量是极其惊人的。
shuffle read 阶段通常就是一个 stage 刚开始时要做的事情。此时该 stage 的每一个 task 就需要将上一个 stage 的计算结果中的所有相同 key从各个节点上通过网络都拉取到自己所在的节点上然后进行 key 的聚合或连接等操作。由于 shuffle write 的过程中map task 给下游 stage 的每个 reduce task 都创建了一个磁盘文件因此 shuffle read 的过程中每个 reduce task 只要从上游 stage 的所有 map task 所在节点上拉取属于自己的那一个磁盘文件即可。
shuffle read 的拉取过程是一边拉取一边进行聚合的。每个 shuffle read task 都会有一个自己的 buffer 缓冲每次都只能拉取与 buffer 缓冲相同大小的数据然后通过内存中的一个 Map 进行聚合等操作。聚合完一批数据后再拉取下一批数据并放到 buffer 缓冲中进行聚合操作。以此类推直到最后将所有数据到拉取完并得到最终的结果。
HashShuffleManager 工作原理如下图所示
优化的HashShuffleManager 为了优化 HashShuffleManager 我们可以设置一个参数spark.shuffle.consolidateFiles该参数默认值为 false将其设置为 true 即可开启优化机制通常来说如果我们使用 HashShuffleManager那么都建议开启这个选项。
开启 consolidate 机制之后在 shuffle write 过程中task 就不是为下游 stage 的每个 task 创建一个磁盘文件了此时会出现 shuffleFileGroup 的概念每个 shuffleFileGroup 会对应一批磁盘文件磁盘文件的数量与下游 stage 的 task 数量是相同的。一个 Executor 上有多少个 cpu core就可以并行执行多少个 task。而第一批并行执行的每个 task 都会创建一个 shuffleFileGroup并将数据写入对应的磁盘文件内。
当 Executor 的 cpu core 执行完一批 task接着执行下一批 task 时下一批 task 就会复用之前已有的 shuffleFileGroup包括其中的磁盘文件也就是说此时 task 会将数据写入已有的磁盘文件中而不会写入新的磁盘文件中。因此consolidate 机制允许不同的 task 复用同一批磁盘文件这样就可以有效将多个 task 的磁盘文件进行一定程度上的合并从而大幅度减少磁盘文件的数量进而提升 shuffle write 的性能。
假设第二个 stage 有 100 个 task第一个 stage 有 50 个 task总共还是有 10 个 ExecutorExecutor CPU 个数为 1每个 Executor 执行 5 个 task。那么原本使用未经优化的 HashShuffleManager 时每个 Executor 会产生 500 个磁盘文件所有 Executor 会产生 5000 个磁盘文件的。但是此时经过优化之后每个 Executor 创建的磁盘文件的数量的计算公式为cpu core的数量 * 下一个stage的task数量也就是说每个 Executor 此时只会创建 100 个磁盘文件所有 Executor 只会创建 1000 个磁盘文件。
优化后的 HashShuffleManager 工作原理如下图所示 优点
可以省略不必要的排序开销。避免了排序所需的内存开销。
缺点
生产的文件过多会对文件系统造成压力。大量小文件的随机读写带来一定的磁盘开销。数据块写入时所需的缓存空间也会随之增加对内存造成压力。
SortShuffle
SortShuffleManager 的运行机制主要分成三种
普通运行机制bypass 运行机制当 shuffle read task 的数量小于等于spark.shuffle.sort.bypassMergeThreshold的值时默认为 200就会启用 bypass 机制Tungsten Sort 运行机制开启此运行机制需设置配置项 spark.shuffle.managertungsten-sort。开启此项配置也不能保证就一定采用此运行机制后面会解释。
普通运行机制
在该模式下数据会先写入一个内存数据结构中此时根据不同的 shuffle 算子可能选用不同的数据结构。如果是 reduceByKey 这种聚合类的 shuffle 算子那么会选用 Map 数据结构一边通过 Map 进行聚合一边写入内存如果是 join 这种普通的 shuffle 算子那么会选用 Array 数据结构直接写入内存。接着每写一条数据进入内存数据结构之后就会判断一下是否达到了某个临界阈值。如果达到临界阈值的话那么就会尝试将内存数据结构中的数据溢写到磁盘然后清空内存数据结构。
在溢写到磁盘文件之前会先根据 key 对内存数据结构中已有的数据进行排序。排序过后会分批将数据写入磁盘文件。默认的 batch 数量是 10000 条也就是说排序好的数据会以每批 1 万条数据的形式分批写入磁盘文件。写入磁盘文件是通过 Java 的 BufferedOutputStream 实现的。BufferedOutputStream 是 Java 的缓冲输出流首先会将数据缓冲在内存中当内存缓冲满溢之后再一次写入磁盘文件中这样可以减少磁盘 IO 次数提升性能。
一个 task 将所有数据写入内存数据结构的过程中会发生多次磁盘溢写操作也就会产生多个临时文件。最后会将之前所有的临时磁盘文件都进行合并这就是merge 过程此时会将之前所有临时磁盘文件中的数据读取出来然后依次写入最终的磁盘文件之中。此外由于一个 task 就只对应一个磁盘文件也就意味着该 task 为下游 stage 的 task 准备的数据都在这一个文件中因此还会单独写一份索引文件 其中标识了下游各个 task 的数据在文件中的 start offset 与 end offset。
SortShuffleManager 由于有一个磁盘文件 merge 的过程因此大大减少了文件数量。比如第一个 stage 有 50 个 task总共有 10 个 Executor每个 Executor 执行 5 个 task而第二个 stage 有 100 个 task。由于每个 task 最终只有一个磁盘文件因此此时每个 Executor 上只有 5 个磁盘文件所有 Executor 只有 50 个磁盘文件。
普通运行机制的 SortShuffleManager 工作原理如下图所示 bypass 运行机制
Reducer 端任务数比较少的情况下基于 Hash Shuffle 实现机制明显比基于 Sort Shuffle 实现机制要快因此基于 Sort huffle 实现机制提供了一个回退方案就是 bypass 运行机制。对于 Reducer 端任务数少于配置属性spark.shuffle.sort.bypassMergeThreshold设置的个数时使用带 Hash 风格的回退计划。
bypass 运行机制的触发条件如下
shuffle map task 数量小于spark.shuffle.sort.bypassMergeThreshold200参数的值。 不是聚合类的 shuffle 算子。 此时每个 task 会为每个下游 task 都创建一个临时磁盘文件并将数据按 key 进行 hash 然后根据 key 的 hash 值将 key 写入对应的磁盘文件之中。当然写入磁盘文件时也是先写入内存缓冲缓冲写满之后再溢写到磁盘文件的。最后同样会将所有临时磁盘文件都合并成一个磁盘文件并创建一个单独的索引文件。
该过程的磁盘写机制其实跟未经优化的 HashShuffleManager 是一模一样的因为都要创建数量惊人的磁盘文件只是在最后会做一个磁盘文件的合并而已。因此少量的最终磁盘文件也让该机制相对未经优化的 HashShuffleManager 来说shuffle read 的性能会更好。
而该机制与普通 SortShuffleManager 运行机制的不同在于第一磁盘写机制不同第二不会进行排序。也就是说启用该机制的最大好处在于shuffle write 过程中不需要进行数据的排序操作也就节省掉了这部分的性能开销。
bypass 运行机制的 SortShuffleManager 工作原理如下图所示 Tungsten Sort Shuffle 运行机制
基于 Tungsten Sort 的 Shuffle 实现机制主要是借助 Tungsten 项目所做的优化来高效处理 Shuffle。
Spark 提供了配置属性用于选择具体的 Shuffle 实现机制但需要说明的是虽然默认情况下 Spark 默认开启的是基于 SortShuffle 实现机制但实际上参考 Shuffle 的框架内核部分可知基于 SortShuffle 的实现机制与基于 Tungsten Sort Shuffle 实现机制都是使用 SortShuffleManager而内部使用的具体的实现机制是通过提供的两个方法进行判断的
对应非基于 Tungsten Sort 时通过 SortShuffleWriter.shouldBypassMergeSort 方法判断是否需要回退到 Hash 风格的 Shuffle 实现机制当该方法返回的条件不满足时则通过 SortShuffleManager.canUseSerializedShuffle 方法判断是否需要采用基于 Tungsten Sort Shuffle 实现机制而当这两个方法返回都为 false即都不满足对应的条件时会自动采用普通运行机制。
因此当设置了 spark.shuffle.managertungsten-sort 时也不能保证就一定采用基于 Tungsten Sort 的 Shuffle 实现机制。
要实现 Tungsten Sort Shuffle 机制需要满足以下条件
Shuffle 依赖中不带聚合操作或没有对输出进行排序的要求。
Shuffle 的序列化器支持序列化值的重定位当前仅支持 KryoSerializer Spark SQL 框架自定义的序列化器。
Shuffle 过程中的输出分区个数少于 16777216 个。
实际上使用过程中还有其他一些限制如引入 Page 形式的内存管理模型后内部单条记录的长度不能超过 128 MB 具体内存模型可以参考 PackedRecordPointer 类。另外分区个数的限制也是该内存模型导致的。
所以目前使用基于 Tungsten Sort Shuffle 实现机制条件还是比较苛刻的。 Spark 底层执行原理 Spark有哪些组件
master管理集群和节点不参与计算。worker计算节点进程本身不参与计算和master汇报。Driver运行程序的main方法创建spark context对象。spark context控制整个application的生命周期包括dagsheduler和task scheduler等组件。client用户提交程序的入口。
Spark 运行流程
具体运行流程如下 构建Application的运行环境Driver创建一个SparkContext SparkContext 向资源管理器Standalone、Mesos、Yarn注册并申请运行 Executor资源管理器分配 Executor然后资源管理器启动 ExecutorExecutor 发送心跳至资源管理器SparkContext 构建 DAG 有向无环图DAGScheduler将DAG图解析成Stage由task Scheduler将Task发送给Executor运行DAGScheduler将DAG图解析成Stage每个Stage有多个task形成TaskSet发送给task SchedulerExecutor 向 SparkContext 申请 TaskTaskScheduler 将 Task 发送给 Executor 运行同时 SparkContext 将应用程序代码发放给 ExecutorTask 在 Executor 上运行运行完毕释放所有资源
从代码角度看 DAG 图的构建
Val lines1 sc.textFile(inputPath1).map(...).map(...)
Val lines2 sc.textFile(inputPath2).map(...)
Val lines3 sc.textFile(inputPath3)
Val dtinone1 lines2.union(lines3)
Val dtinone lines1.join(dtinone1)
dtinone.saveAsTextFile(...)
dtinone.filter(...).foreach(...)
上述代码的 DAG 图如下所示 Spark 内核会在需要计算发生的时刻绘制一张关于计算路径的有向无环图也就是如上图所示的 DAG。Spark 的计算发生在 RDD 的 Action 操作而对 Action 之前的所有 TransformationSpark 只是记录下 RDD 生成的轨迹而不会触发真正的计算。
将 DAG 划分为 Stage 核心算法
一个 Application 可以有多个 job 多个 Stage
Spark Application 中可以因为不同的 Action 触发众多的 job一个 Application 中可以有很多的 job每个 job 是由一个或者多个 Stage 构成的后面的 Stage 依赖于前面的 Stage也就是说只有前面依赖的 Stage 计算完毕后后面的 Stage 才会运行。
划分依据Stage 划分的依据就是宽依赖像 reduceByKeygroupByKey 等算子会导致宽依赖的产生。
核心算法回溯算法
从后往前回溯/反向解析遇到窄依赖加入本 Stage遇见宽依赖进行 Stage 切分。
Spark 内核会从触发 Action 操作的那个 RDD 开始从后往前推首先会为最后一个 RDD 创建一个 Stage然后继续倒推如果发现对某个 RDD 是宽依赖那么就会将宽依赖的那个 RDD 创建一个新的 Stage那个 RDD 就是新的 Stage 的最后一个 RDD。然后依次类推继续倒推根据窄依赖或者宽依赖进行 Stage 的划分直到所有的 RDD 全部遍历完成为止。
将 DAG 划分为 Stage 剖析 一个 Spark 程序可以有多个 DAG(有几个 Action就有几个 DAG上图最后只有一个 Action图中未表现,那么就是一个 DAG)。
一个 DAG 可以有多个 Stage(根据宽依赖/shuffle 进行划分)。
同一个 Stage 可以有多个 Task 并行执行(task 数分区数如上图Stage1 中有三个分区 P1、P2、P3对应的也有三个 Task)。
可以看到这个 DAG 中只 reduceByKey 操作是一个宽依赖Spark 内核会以此为边界将其前后划分成不同的 Stage。
同时我们可以注意到在图中 Stage1 中从 textFile 到 flatMap 到 map 都是窄依赖这几步操作可以形成一个流水线操作通过 flatMap 操作生成的 partition 可以不用等待整个 RDD 计算结束而是继续进行 map 操作这样大大提高了计算的效率。
提交 Stages
调度阶段的提交最终会被转换成一个任务集的提交DAGScheduler 通过 TaskScheduler 接口提交任务集这个任务集最终会触发 TaskScheduler 构建一个 TaskSetManager 的实例来管理这个任务集的生命周期对于 DAGScheduler 来说提交调度阶段的工作到此就完成了。
而 TaskScheduler 的具体实现则会在得到计算资源的时候进一步通过 TaskSetManager 调度具体的任务到对应的 Executor 节点上进行运算。 任务调度总体诠释 补充一、scala 脚本 模版
SourceTable.scala
trait SourceTable {val BASIC_INFO basic_info
}
OutputTable.scala
trait OutputTable {val TARGET_TABLE target_table
}
GeneratorDataTask.scala
object GeneratorDataTask extend SourceTable with OutputTable {private val argsKeys Set(dt, env)val INDEX_DT: String DuccUtil.getLastDt(last_dt);def main(args: Array[String]): Unit {val spark SparkSessionUtil.getSession(GeneratorDataTask)try {val properties TaskUtil.getArgs(args, argsKeys)if (!properties.contains(dt) || !properties.contains(env)) {throw new Exception(必要参数为空)}val selectDF spark.read.table(BASIC_INFO).where(sdt $INDEX_DT) selectDF.na.fill(0).write.mode(SaveMode.Overwrite).insertInto(TARGET_TABLE) } catch {case e: Exception {println(Excetion detail:, e)System.exit(1)}} finally {spark.stop() }}
补充二、Spark任务运行流程
简单的说 用户在client端提交作业后会由Driver运行main方法并创建spark context上下文。调用RDD上的方法形成dag图输入dagscheduler按照rdd之间的依赖关系划分stage输入task scheduler。task scheduler会将stage划分为task set分发到各个节点的executor中执行。 再具体的说
客户端提交作业Driver启动流程Driver申请资源并启动其余Executor(即Container)Executor启动流程作业调度生成stages与tasks。Task调度到Executor上Executor启动线程执行Task逻辑Driver管理Task状态Task完成Stage完成作业完成
在工厂环境下Spark 集群的部署方式一般为 YARN-Cluster 模式 时序图 提交一个Spark应用程序首先通过Client向 ResourceManager 请求启动一个Application同时检查是否有足够的资源满足 Application 的需求如果资源条件满足则准备 ApplicationMaster的启动上下文交给ResourceManager并循环监控Application状态。当提交的资源队列中有资源时ResourceManager 会在某个 NodeManager 上启动 ApplicationMaster 进程ApplicationMaster 会单独启动 Driver 后台线程当 Driver启动后ApplicationMaster 会通过本地的 RPC 连接 Driver并开始向 ResourceManager申请 Container 资源运行 Executor 进程一个 Executor 对应与一个 Container当ResourceManager 返回 Container 资源ApplicationMaster 则在对应的 Container 上启动 Executor。Driver 线程主要是初始化 SparkContext 对象准备运行所需的上下文然后一方面保持与 ApplicationMaster 的 RPC 连接通过 ApplicationMaster 申请资源另一方面根据用户业务逻辑开始调度任务将任务下发到已有的空闲 Executor 上。Driver 初始化 SparkContext 过 程 中 会 分 别 初 始 化 DAGScheduler 、TaskScheduler、SchedulerBackend 以及 HeartbeatReceiver并启动 SchedulerBackend以及 HeartbeatReceiver。SchedulerBackend 通过 ApplicationMaster 申请资源并不断从 TaskScheduler 中拿到合适的 Task 分发到 Executor 执行。HeartbeatReceiver 负责接收 Executor 的心跳信息监控 Executor 的存活状况并通知到 TaskScheduler。当 ResourceManager 向 ApplicationMaster 返回 Container 资源时ApplicationMaster 就尝试在对应的 Container 上启动 Executor 进程Executor 进程起来后会向 Driver 反向注册注册成功后保持与 Driver 的心跳同时等待 Driver分发任务当分发的任务执行完毕后将任务状态上报给 Driver。
从任务调度的阶段来看四个步骤
1.构建DAG调用RDD上的方法2.DAGScheduler将DAG切分Stage切分的依据是Shuffle将Stage中生成的Task以TaskSet的形式给TaskScheduler3.TaskScheduler调度Task根据资源情况将Task调度到相应的Executor中4.Executor接收Task然后将Task丢入到线程池中执行 补充三、为什么说与MapReduce相比Spark运行效率更高
reduceByKey的结果是一个新的RDD其中每个键都唯一与每个键相关联的值经过了聚合操作。
groupByKey的结果是一个新的RDD其中每个键都与一个迭代器相关联迭代器包含了与该键关联的所有值。
spark借鉴了Mapreduce继承了其分布式计算的优点并进行了改进spark生态更为丰富功能更为强大性能更加适用范围广mapreduce更简单稳定性好。主要区别
1spark把运算的中间数据(shuffle阶段产生的数据)存放在内存迭代计算效率更高mapreduce的中间结果需要落地保存到磁盘
2Spark容错性高它通过弹性分布式数据集RDD来实现高效容错RDD是一组分布式的存储在 节点内存中的只读性的数据集这些集合石弹性的某一部分丢失或者出错可以通过整个数据集的计算流程的血缘关系来实现重建mapreduce的容错只能重新计算
3Spark更通用提供了transformation和action这两大类的多功能api另外还有流式处理spark streaming模块、图计算等等mapreduce只提供了map和reduce两种操作流计算及其他的模块支持比较缺乏
4Spark框架和生态更为复杂有RDD血缘lineage、执行时的有向无环图DAG,stage划分等很多时候spark作业都需要根据不同业务场景的需要进行调优以达到性能要求mapreduce框架及其生态相对较为简单对性能的要求也相对较弱运行较为稳定适合长期后台运行。
5Spark计算框架对内存的利用和运行的并行度比mapreduce高Spark运行容器为executor内部ThreadPool中线程运行一个Task,mapreduce在线程内部运行containercontainer容器分类为MapTask和ReduceTask.程序运行并行度高
6Spark对于executor的优化在JVM虚拟机的基础上对内存弹性利用storage memory与Execution memory的弹性扩容使得内存利用效率更高 补充四、RDD中reduceBykey与groupByKey哪个性能好
reduceByKey是一个转换操作transformation它会对具有相同键的值进行聚合操作。这个操作会遍历数据集中的每个键值对并将具有相同键的值传递给一个减少函数reduce function该函数会将这些值组合成一个单一的值。示例
from pyspark import SparkContextsc SparkContext(local, ReduceByKey Example)data [(a, 1), (b, 1), (a, 1), (a, 1), (b, 1), (c, 1)]
rdd sc.parallelize(data)# 使用reduceByKey进行聚合
result rdd.reduceByKey(lambda x, y: x y)print(result.collect())
# 输出: [(a, 3), (b, 2), (c, 1)]
groupByKey也是一个转换操作它会将具有相同键的所有值组合在一起并返回一个键值对其中键是原始键值是一个迭代器包含了所有具有该键的值。 示例
from pyspark import SparkContextsc SparkContext(local, GroupByKey Example)data [(a, 1), (b, 1), (a, 1), (a, 1), (b, 1), (c, 1)]
rdd sc.parallelize(data)# 使用groupByKey进行分组
grouped rdd.groupByKey()# 对分组后的结果进行转换以便查看
result grouped.mapValues(lambda values: list(values))print(result.collect())
# 输出: [(a, [1, 1, 1]), (b, [1, 1]), (c, [1])]
reduceByKeyreduceByKey会在结果发送至reducer之前会对每个mapper在本地进行merge有点类似于在MapReduce中的combiner。这样做的好处在于在map端进行一次reduce之后数据量会大幅度减小从而减小传输保证reduce端能够更快的进行结果计算。
groupByKeygroupByKey会对每一个RDD中的value值进行聚合形成一个序列(Iterator)此操作发生在reduce端所以势必会将所有的数据通过网络进行传输造成不必要的浪费。同时如果数据量十分大可能还会造成OutOfMemoryError。
所以在进行大量数据的reduce操作时候建议使用reduceByKey。不仅可以提高速度还可以防止使用groupByKey造成的内存溢出问题。 补充五、Shuffle数据块有多少种不同的存储方式
RDD数据块用来存储所缓存的RDD数据。Shuffle数据块用来存储持久化的Shuffle数据。广播变量数据块用来存储所存储的广播变量数据。任务返回结果数据块用来存储在存储管理模块内部的任务返回结果。通常情况下任务返回结果随任务一起通过Akka返回到Driver端。但是当任务返回结果很大时会引起Akka帧溢出这时的另一种方案是将返回结果以块的形式放入存储管理模块然后在Driver端获取该数据块即可因为存储管理模块内部数据块的传输是通过Socket连接的因此就不会出现Akka帧溢出了。流式数据块只用在Spark Streaming中用来存储所接收到的流式数据块
补充六、Scala里trait有什么功能与class有何异同什么时候用trait什么时候该用class
trait可以被继承而且支持多重继承其实它更像我们熟悉的接口interface但它与接口又有不同之处是trait中可以写方法的实现interface不可以java8开始支持接口中允许写方法实现代码了这样看起来trait又很像抽象类。
补充七、Scala 语法中to 和 until有啥区别
to 包含上界until不包含上界。
补充八、Scala伴生对象和伴生类
单例对象与类同名时这个单例对象被称为这个类的伴生对象而这个类被称为这个单例对象的伴生类。伴生类和伴生对象要在同一个源文件中定义伴生对象和伴生类可以互相访问其私有成员。不与伴生类同名的单例对象称为孤立对象。
import scala.collection.mutable.Mapclass ChecksumAccumulator {private var sum 0def add(b: Byte) {sum b}def checksum(): Int ~(sum 0xFF) 1
}object ChecksumAccumulator {private val cache Map[String, Int]()def calculate(s: String): Int if (cache.contains(s))cache(s)else {val acc new ChecksumAccumulatorfor (c - s)acc.add(c.toByte)val cs acc.checksum()cache (s - cs)println(s:s cs:cs)cs}def main(args: Array[String]) {println(Java 1:calculate(Java))println(Java 2:calculate(Java))println(Scala :calculate(Scala))}
}补充九、Spark 中 PartitionTaskcoreExecutor的个数决定因素和关系
Partition 是 Spark RDD 计算的最小单元决定了计算的并发度。分区数如果远小于集群可用的 CPU 数不利于发挥 Spark 的性能还容易导致数据倾斜等问题。分区数如果远大于集群可用的 CPU 数会导致资源分配的时间过长从而影响性能。
默认情况下Task 的数量是由 Partition 决定的RDD 计算时每个分区会启一个 Task所以 RDD 的分区数决定了总 Task 数。
补充十、介绍一下join操作优化经验
join其实常见的就分为两类 map join 和 reduce join。
当大表和小表join时用map join能显著提高效率。将多份数据进行关联是数据处理过程中非常普遍的用法不过在分布式计算系统中这个问题往往会变的非常麻烦因为框架提供的 join 操作一般会将所有数据根据 key 发送到所有的 reduce 分区中去也就是 shuffle 的过程。造成大量的网络以及磁盘IO消耗运行效率极其低下这个过程一般被称为 reduce join。如果其中有张表较小的话我们则可以自己实现在 map 端实现数据关联跳过大量数据进行 shuffle 的过程运行时间得到大量缩短根据不同数据可能会有几倍到数十倍的性能提升。
原则就是尽量不进行reduce join。
两个大表之间进行join操作影响性能的主要因素是数据倾斜我们要进行尽量保证join的两张表发送到executor的数据的数量是一样的而这个可以通过distributed by join(条件列)进行这样可以提前把两个表的数据按照条件列分布好在进行join操作时就不会发生数据倾斜的问题了
ps: distributed by 条件列 是把数据按照条件列进行分区分区的数量由set spark.sql.shuffle.partitions600; 进行控制此外即使不是用于join操作遇到表数据倾斜是我们也可以使用例如select * from Table distribute by rand(); 这样就可以保证每个分区的数据基本一致了。 补充十一、Spark的数据本地性有哪几种
Spark中的数据本地性有三种
1PROCESS_LOCAL是指读取缓存在本地节点的数据
2NODE_LOCAL是指读取本地节点硬盘数据
3ANY是指读取非本地节点数据
通常读取数据PROCESS_LOCALNODE_LOCALANY尽量使数据以PROCESS_LOCAL或NODE_LOCAL方式读取。其中PROCESS_LOCAL还和cache有关如果RDD经常用的话将该RDD cache到内存中注意由于cache是lazy的所以必须通过一个action的触发才能真正的将该RDD cache到内存中。
补充十二、scala代码实现WordCount
val conf new SparkConf()
val sc new SparkContext(conf)
val line sc.textFile(xxxx.txt)
line.flatMap(_.split( )).map((_,1)).reduceByKey(__). collect().foreach(println)
sc.stop()
