内容简介：Spark Learning Note Part 2
Components
首先我们了解一下spark各个组件部分，及其分工。
Spark Core
Spark Core是spark的核心模块。我的理解是这个部分去实现了基本的功能，是底层的支持部分，其他组件会对其进行扩展、继承和调用。具体完成了以下功能:
- 任务调度
- 内存管理
- 错误恢复
- 和存储系统交互
Spark SQL
Spark SQL是用于操作结构化数据的程序包，支持多种数据源，比如hive表、Parquet以及JSON等。除了提供一个 SQL 接口，还支持将SQL和传统的RDD编程的数据操作方式结合。
Spark SQL是在Spark 1.0中被引入的。
Spark Streaming
Spark Streaming是针对实时数据进行流式计算的组件。Spark Streaming提供了用于操作数据流的API，并且与Spark Core中的RDD API对应。
MLlib
MLlib提供了机器学习的算法，包括分类、回归、聚类、协同过滤等。
GraphX
GraphX用于操作图的程序库，进行并行的图计算。
集群调度器
- 独立调度器
- Hadoop Yarn
- Apache Mesos ( learn more )
RDD Programming
RDD(Resilient Distributed Dataset)即弹性分布式数据集。在spark中，对数据的所有操作即创建RDD、转换RDD、对RDD求值。
RDD基础
用户可以使用两种方法创建RDD:
- 读取一个外部数据集
lines = sc.textFile("README.md")
RDD支持两种类型的操作: transformation 以及 action。
transformation操作会由一个RDD生成一个新的RDD，如filter()操作。
pythonLines = lines.filter(lambda line: "python" in line)
action会对RDD计算出一个结果，并把结果返回到驱动程序中或存储到外部系统中。如filter()操作。
pythonLines.first()
值得注意的是Spark采用惰性计算RDD的策略。什么时候开始计算及这样做的目的是什么?
- RDD第一次在一个行动操作中用到时，才会计算。
- Spark偏向于了解完整的转化操作链之后，便于只操作需要结果的数据集。(以上面的文件扫描为例，当知道只要第一条数据[first()]，那么在进行sc.textFile(“README.md”)过程时，只要存储第一条数据即可。)
另一点值得注意的是，RDD会在每次对其进行行动操作时进行重新计算，若想在多个行动操作中重用一个RDD，可以使用RDD.persist()让RDD缓存下来。(Spark会将RDD的内容保存到内存中，以分区的方式存储到集群中的各机器上) 那么为什么默认不对RDD进行持久化呢?
- 若数据规模很大，全部占用内存会很消耗资源，所以人为对需要存储的部分进行一次筛选会更好。
创建RDD
创建RDD的方式:
- 读取外部数据集
- 驱动器程序中对一个集合进行并行化
lines = sc.parallelize(["pandas", "i like pandas"])
此方式常用于开发原型和测试中。
向Spark传递函数
以 python 为例
- lambda
word = rdd.filter(lambda s: "error" in s)
- Py function
def containsError(s):
return "error" in s
word = rdd.filter(containsError)
- 值得注意的是，在函数传递的实现过程中，要避免整个对象的序列化传递。
class SearchFunctions(object):
def __init__(self, query):
self.query = query
def getMatchesMemberReference(self, rdd):
query = self.query (It is necessary to do assignment)
return rdd.filter(lambda x: query in x)
常见的转化操作和行动操作
- 针对各个元素的转化操作
(1) map()
接收一个函数，将函数用于RDD中的每个元素，将函数的返回结果作为结果RDD中对应元素的值。
nums = sc.parallelize([1,2,3,4])
squared = nums.map(lambda x: x * x).collect() (rdd to collection)
for num in squared:
print num
compare with flatMap()
(2) filter()
接收一个函数，并将RDD中满足该函数的元素放入新的RDD中返回。
nums = sc.parallelize([1,2,3,4]) result = nums.filter(lambda x: x>2).collect()
- 伪集合操作
1. RDD.distinct()
生成一个只包含不同元素的新RDD。该操作的开销很大，需要通过网络进行shuffle。
2. RDD1.union(RDD2)
返回包含RDD1, RDD2的所有元素集合。
3. RDD1.intersection(RDD2)
返回包含RDD1, RDD2的所有元素集合, 并去除重复部分。性能相较于union会差，因为要经过唯一性的shuffle()。
4. RDD1.substract(RDD2)
返回只存在于RDD1而不存在于RDD2的元素组成的RDD，需要shuffle。
5. RDD1.cartesian(RDD2)
- 行动操作
1. reduce()
接收一个函数作为参数，函数会操作两个RDD的元素类型的数据并返回一个同样类型的新元素。
sum = rdd.reduce(lambda x, y: x+y)
2. foreach()
比如以json格式把数据发送到一个网络服务器上，或者把数据存到数据库中。
持久化
以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网
猜你喜欢:
本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。
