Spark Learning Note Part 2

Components

首先我们了解一下spark各个组件部分,及其分工。

Screen Shot 2017-06-17 at 10.54.17 PM

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()

Screen Shot 2017-06-17 at 10.53.57 PM

(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)

Screen Shot 2017-06-17 at 10.53.47 PM

  • 行动操作

1. reduce()

接收一个函数作为参数,函数会操作两个RDD的元素类型的数据并返回一个同样类型的新元素。

sum = rdd.reduce(lambda x, y: x+y)

2. foreach()

比如以json格式把数据发送到一个网络服务器上,或者把数据存到数据库中。

持久化

Screen Shot 2017-06-17 at 10.53.29 PM

 

Refer to:

Learning Spark: Lightning-fast Data Analysis

 

 

 

 

 

发表评论

电子邮件地址不会被公开。 必填项已用*标注

您可以使用这些HTML标签和属性: <a href="" title=""> <abbr title=""> <acronym title=""> <b> <blockquote cite=""> <cite> <code> <del datetime=""> <em> <i> <q cite=""> <strike> <strong>