Spark Learning Note Part 2

内容简介：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格式把数据发送到一个网络服务器上，或者把数据存到数据库中。

持久化

以上就是本文的全部内容，希望本文的内容对大家的学习或者工作能带来一定的帮助，也希望大家多多支持 码农网

查看所有标签

本站部分资源来源于网络，本站转载出于传递更多信息之目的，版权归原作者或者来源机构所有，如转载稿涉及版权问题，请联系我们。