快速大数据分析

图 灵 程 序 设 计 丛 书

人 民 邮 电 出 版 社

北　　京

Learning Spark

Lightning-Fast Data Analysis

[美] Holden Karau [美] Andy Konwinski [美] Patrick Wendell [加] Matei Zaharia 著

王道远 译 英特尔大数据技术中心 审校

Spark快速大数据分析

Beijing • Cambridge • Farnham • Köln • Sebastopol • Tokyo

O’Reilly Media, Inc.授权人民邮电出版社出版

内 容 提 要

本书由Spark 开发者及核心成员共同打造，讲解了网络大数据时代应运而生的、能高效迅

捷地分析处理数据的工具——Spark，它带领读者快速掌握用 Spark 收集、计算、简化和保存海 量数据的方法，学会交互、迭代和增量式分析，解决分区、数据本地化和自定义序列化等问题。

本书适合大数据时代所有需要进行数据分析的人员阅读。

定价：59.00元

读者服务热线：(010)51095186转600　印装质量热线：(010)81055316 反盗版热线：(010)81055315

广告经营许可证：京崇工商广字第 0021 号 著　　　　[美] Holden Karau Andy Konwinski 　　　　　　　 Patrick Wendell

　　　　　[加] Matei Zaharia 译　　　　王道远

审　　校　英特尔大数据技术中心 责任编辑　岳新欣

执行编辑　张　曼 责任印制　杨林杰

人民邮电出版社出版发行　　北京市丰台区成寿寺路11号

邮编　100164　　电子邮件　315@ptpress.com.cn 网址　http://www.ptpress.com.cn

北京　　　　　　印刷 开本：800×1000　1/16 印张：14.513

字数：343千字 2015年 9 月第 1 版

印数：1 — 3 500册 2015年 9 月北京第 1次印刷

著作权合同登记号 图字：01-2015-3678号

◆

◆

◆

版权声明

© 2015 by O’Reilly Media, Inc.

Simplified Chinese Edition, jointly published by O’Reilly Media, Inc. and Posts & Telecom Press, 2015. Authorized translation of the English edition, 2015 O’Reilly Media, Inc., the owner of all rights to publish and sell the same.

All rights reserved including the rights of reproduction in whole or in part in any form.

英 版由 O’Reilly Media, Inc. 出版，2015。

简 中 版由人民邮电出版社出版， 2015。英 版的 译 O’Reilly Media, Inc. 的 权。 简 中 版的出版和 出版权和 权的所有者——O’Reilly Media, Inc.

的 。

版权所有， 书 ，本书的任 分和 任 式 制。

O’Reilly Media 图书、 、 、 和会 等方式 新 。

自 1978 年开 ，O’Reilly 发 的 者和 者。 开

，而 要的技术 —— 大 的 号 社

会 新 技的应用。作 技术社区中 的 者，O’Reilly 的发 了 新的

、 造和发 大。

O’Reilly 件开发人员带 的 书 第 网 GNN

了 远的开 代 会， 开 件运 了 Make ，

而成 DIY 的 要 地 式 人的 带。

O’Reilly 的会 和 会集 了 和高 远 的 领 ，共同 出开

新 的 。作 技术人 的 ，O’Reilly 的

的计算 用 。 书 出版、 者 ，

O’Reilly 的 了 的理 —— 发 新的 量。

O’Reilly Radar

Wired O’Reilly

Business 2.0 O’Reilly Conference

CRN O’Reilly

Irish Times

Tim 人

Yogi Berra Tim

Linux Journal

O’Reilly Media, Inc.介绍

目录

...xi

...xiv

...xvi

...xvii

1 Spark 数据分析

1.1　Spark ...1

1.2　 大 的 件 ...2

1.2.1 Spark Core ...2

1.2.2 Spark SQL...3

1.2.3 Spark Streaming ...3

1.2.4 MLlib ...3

1.2.5 GraphX...3

1.2.6 ...4

1.3　Spark 的用 和用 ...4

1.3.1 ...4

1.3.2 ...5

1.4　Spark 简 ...5

1.5　Spark 的版本和发 ...6

1.6　Spark 的存 次...6

2 Spark

2.1　 Spark...7

2.2　Spark 中 Python 和 Scala 的 shell ...9

2.3　Spark 核心 简 ...12

2.4　 应用 ...14

2.4.1 SparkContext ...15

2.4.2 ...16

2.5　 ...19

3 RDD

3.1　RDD ...21

3.2　 RDD ...23

3.3　RDD 作 ...24

3.3.1 ...24

3.3.2 ...26

3.3.3 ...27

3.4　 Spark 数...27

3.4.1 Python ...27

3.4.2 Scala ...28

3.4.3 Java ...29

3.5　 的 化 作和行 作 ...30

3.5.1 RDD ...30

3.5.2 RDD ...37

3.6　 化( 存) ...39

3.7　 ...40

4

4.1　 ...41

4.2　 Pair RDD ...42

4.3　Pair RDD 的 化 作 ...42

4.3.1 ...45

4.3.2 ...49

4.3.3 ...50

4.3.4 ...51

4.4　Pair RDD 的行 作 ...52

4.5　数据分区 进 ...52

4.5.1 RDD ...55

4.5.2 ...56

4.5.3 ...57

4.5.4 PageRank ...57

4.5.5 ...59

4.6　 ...61

5 数据

5.1　 ...63

5.2　 件 式 ...64

5.2.1 ...64

5.2.2 JSON ...66

5.2.3 ...68

5.2.4 SequenceFile ...71

5.2.5 ...73

5.2.6 Hadoop 出 ...73

5.2.7 ...77

5.3　 件 ...78

5.3.1 / ...78

5.3.2 Amazon S3 ...78

5.3.3 HDFS ...79

5.4　Spark SQL 中的 化数据 ...79

5.4.1 Apache Hive ...80

5.4.2 JSON ...80

5.5　数据 ...81

5.5.1 Java ...81

5.5.2 Cassandra ...82

5.5.3 HBase ...84

5.5.4 Elasticsearch ...85

5.6　 ...86

6 Spark

6.1　简 ...87

6.2　 加 ...88

6.2.1 ...90

6.2.2 ...91

6.3　 量 ...91

6.4　 分区进行 作 ...94

6.5　 序 的 道 ...96

6.6　数 RDD 的 作 ...99

6.7　 ...100

7 Spark

7.1　简 ...101

7.2　Spark 运行时 ...101

7.2.1 ...102

7.2.2 ...103

7.2.3 ...103

7.2.4 ...104

7.2.5 ...104

7.3　 用spark-submit 应用 ...105

7.4　打 代 ...107

7.4.1 Maven Java Spark ...108

7.4.2 sbt Scala Spark ...109

7.4.3 ... 111

7.5　Spark 应用 应用 ...111

7.6　集 理 ...112

7.6.1 ...112

7.6.2 Hadoop YARN ... 115

7.6.3 Apache Mesos ... 116

7.6.4 Amazon EC2 ... 117

7.7　 合适的集 理 ...120

7.8　 ...121

8 Spark

8.1　 用 SparkConf Spark ...123

8.2　Spark 执行的 成 分：作 、任 和 ...127

8.3　 ...131

8.3.1 Spark ...131

8.3.2 ...134

8.4　 能 量 ...135

8.4.1 ...135

8.4.2 ...136

8.4.3 ...137

8.4.4 ...138

8.5　 ...139

9 Spark SQL

9.1　 Spark SQL ...142

9.2　 应用中 用 Spark SQL ...144

9.2.1 Spark SQL...144

9.2.2 ...145

9.2.3 SchemaRDD ...146

9.2.4 ...148

9.3　读 和存 数据 ...149

9.3.1 Apache Hive ...149

9.3.2 Parquet ...150

9.3.3 JSON ...150

9.3.4 RDD ...152

9.4　JDBC/ODBC ...153

9.4.1 Beeline ...155

9.4.2 ...156

9.5　用 自定义 数 ...156

9.5.1 Spark SQL UDF ...156

9.5.2 Hive UDF ...157

9.6　Spark SQL 能 ...158

9.7　 ...159

10 Spark Streaming

10.1　 简 的 子 ...162

10.2　 ...164

10.3　 化 作 ...167

10.3.1 ...167

10.3.2 ...169

10.4　 出 作 ...173

10.5　 ...175

10.5.1 ...175

10.5.2 ...176

10.5.3 ...179

10.6　24/7 运行 ...180

10.6.1 ...180

10.6.2 ...181

10.6.3 ...182

10.6.4 ...182

10.6.5 ...183

10.7　Streaming 用 ...183

10.8　 能 量 ...184

10.8.1 ...184

10.8.2 ...184

10.8.3 ...185

10.9　 ...185

11 MLlib

11.1　 ...187

11.2　 要 ...188

11.3　 学 ...189

11.4　数据 ...192

11.5　算法 ...194

11.5.1 ...194

11.5.2 ...196

11.5.3 ...197

11.5.4 ...202

11.5.5 ...203

11.5.6 ...204

11.5.7 ...206

11.6　 能 量 ...206

11.6.1 ...206

11.6.2 ...207

11.6.3 RDD ...207

11.6.4 ...207

11.6.5 ...207

11.7　 API ...208

11.8　 ...209

...210

...210

推荐序

年 大数据 ， 有人问 大数据 要。 处 数据 的时代，

大量 的 能 、 、 及 网 时 生新的数据。

，有 90% 的数据 年 生的。 2020 年， 有 500 台的互

生 Zeta 字 的数据。带 的 海量数据本 ，而 用

数据。大数据解决方 的 大 它 快速处理大 、 的数据集，

方法 快、 地生成 。

大数据解决方 要 件， 存 、计算和网络等 件 ， 数据处理

， 用 的 计和计算算法、数据 化 的分析 。 中 ，

数据处理 了 分 要的作用。 张地 ，数据处理 大数据 CPU

计算 ， 大 人 。

2009 年，Matei Zaharia 加 大学 分校的 AMPLab 进行 时 了 Spark 大数据处理和计算 。 同 的数据处理 ，Spark 存的 本

primitive 应用 序带 了 100 的 能 。Spark 用 序 数据加

集 存中用 ， 适用 大数据和 学 ， 成 用的大数据

。 Cloudera 和 MapR 的大数据发行版 发 时 加了 Spark。

，Spark Hadoop 和大数据生 发生 ， 地 的大数

据分析需 ， ：Spark Spark 核心，发 了 Spark streaming、SQL、MLlib、

GraphX、SparkR 等 。学 Spark 和它的 件 有 大数据处理速

， 能 开发者和数据 学 地 分析应用。 、 、交

，Spark 的大数据解决方 所 的 量 进 的 成，带

的 加速决 制定。

年中， 的 有 会 本书的作者合作， Apache Spark 社区 成 ，

英特尔 化 大数据和 Spark 应用。 Spark 快速大数据分析 的出版 开发者和

数据 学 了丰 的 Spark 。 要的 ， 本书 简 地 开发者 用

Spark，而 了 Spark 的 成， 了 化大数据应

用。 大 本书， 具 ， 本书 的 化方法和 路， 它

能 出 的大数据应用。

2015 7

Big data is getting hot in recent years. Quite often, folks ask why big data is a big deal. We are in the era of data explosion, with the emergence of smart phones, tablets, wearables, IoT devices, etc. Ninety percent of the data in the world today was generated in just the past two years. By 2020, we will see >50B devices connected and Zeta byte data created. It is not the quantity of the data that is revolutionary. It is that we can now do something with it that's revolutionary. The power of big data solutions is they can process large and complex data sets very fast, generate better and faster insights than conventional methods.

A big data solution suite can consist of several critical components, from the hardware layer like storage, compute and network, to data processing engine, to analytics layer where business insights are generated using improved statistical & computational algorithms and data visualization. Among all, the data processing engine is one most critical player. It is not over- stating that the data processing engine for big data is like CPU for a computer or brain for a human being.

Spark was initially started for the purpose of creating a big data processing and computing framework, when Matei Zaharia was doing his Ph.D. research at UC Berkeley AMPLab in 2009.

Different from the traditional data processing framework, Spark's in-memory primitives provide performance up to 100 times faster for certain applications. By allowing user programs to load data into a cluster's memory and query it repeatedly, Spark is well-suited for big data and machine learning use cases. Spark is becoming one best adopted among all big data modules. Big Data Distributions like Cloudera, MapR now all include Spark into their distributions.

Spark is now evolving the Hadoop and big data ecosystem to better support the end-to-end

big data analytics needs, e.g. Spark grew beyond Spark core to Spark streaming, SQL, MLlib,

GraphX, SparkR, etc. Learning Spark and its internals will not just help improve the processing

speed for big data, but also help developers and data scientists create analytics applications with

more ease. With big data solutions like Spark, we expect to see significant improvement with

business insights which will help expedite the decision making—like we've never seen before, from enterprise, healthcare, transportation, and retail.

Over the years, my organization had the opportunities to work with authors of this book, contribute to Apache Spark, and optimize various Big Data and Spark application on Intel Architecture. The publication of Learning Spark offers developers and data scientists extensive knowledge on Spark. Moreover, Learning Spark does not simply try to tell the developers how to use Spark, it also addresses the internals and shows various examples of how to improve your big data applications. I recommend Learning Spark—that this book, and, more specifically, the method it espouses, will change your big data application for the better.

Ziya Ma, General Manager of the global Big Data Technologies organization, SSG STO, Intel Corp.

Santa Clara, California, July 2015

译者序

大数据 年 的 。 ，互 网 发 ， 生 的

， 而 的 数据量的 数 增 。 ，人 学会了 分析数据 有

价 的 。有时， 的 ， 的数据 法有效保 所有 的 ，

出的 有效。 用 ，而 数据 大， 需要 进数据处理

的 。 人工 计 用 的计算 件进行分析， MapReduce ，

数据 增 ， 处理数据的方式 。 ， 件 的 发

存计算成 了 能， Spark 由 出 ， 它的 字 ， ，迅速 了

工 的 。

Spark 快速大数据分析 本 Spark 学者 的书，它 有 ，而

用 的具 用法。 ，本书 Spark 的用法，它 Spark 的

核心 和 本 理 有 的 ， 读者能 所 。

Spark 用计算 ， 用 Spark 的应用 价 所 。 欣

地 ， 的 互 网 用 Spark 造出了 量的价 。本书的读

者 Spark 应用 中， 数据海 的 。

本书 成， 开 方 。 人民邮电出版社图 的 、岳新欣

、张曼 ， 本译 的出版 了大 。 本人所 的英特尔

发有 大数据 ， 中 、 、 、张 分 责了本书 分的审校

工作， 、 、 本书的 译工作 了 。 Databricks 的 学

， 成了 出版社的合作。 译的 中， 自 人 的理解和

。

本书所 ，Spark 大 的 件 ， 及方方 的 ， 本书的 译增加了

。 译者 保 译的 ，由 学 有 ， 会有 处。而

大数据作 新 学 ， 术 有 定 成的译法。Spark 发 中，本

书英 据 Spark 1.2 编 ，而译者 量 了 Spark 1.4 译时的 新

版本 的 化。 读者发 了本书中的 处， 。 的

电子邮 ：me@daoyuan.wang。

2015

序

Spark 作 代大数据处理 ， 的时 ，

。Spark 大数据 的 Hadoop MapReduce 的 进 要

方 ： ， Spark 速 快 次， Spark 丰 的 API 带 了 大的 用 ，

Spark 处理应用， 交互式 、 式计算、 学 、图计算等

应用， 同应用 的需 。

能 地 Spark 的开发中， Spark 路 ， Spark

的 成 的大数据开 。 ， Spark 成 Apache 会

的 。 ， Spark 人 Matei Zaharia 及 Spark 开发者 Patrick Wendell、Andy Konwinski 和 Holden Karau 由 高 。

成了本书的著作工作。

Spark 的迅速 行， 的问题 时 出 。本书共有 11 ，

学 Spark 的数据 学 、学生、开发者 计的具 ，大大 解了

Spark 的问题。 有大数据方 的读者， 本书

作 大数据领 的 。 地 本书能 领 和 读者 进大数据

人 的新领 ， 年 。

Databricks AMPlab Ion Stoica

前言

行数据分析 行， 行 的工作者 需要 的数据分析工具。

Spark 应运而生， 迅速 了 。作 MapReduce 的 者，Spark 要有 。

， Spark 用。由 高 API 了 集 本 的 ， 所要

的计算本 ， 需 自 的 记本电 开发 Spark 应用。 次，Spark 快，

交互式 用和 算法。 ， Spark 用 ， 用它 成 的运算，

SQL 、 本处理、 学 等，而 Spark 出 ， 需要学

的 分 处理 需 。 大 Spark 作 学 大数据的

的 。

本书 要 Spark， 读者能 Spark。 能 本书中学 Spark

的电 运行 ， 交互式 作 学 Spark 的 API。 会讲解 用 Spark 作数据 作和分 式执行时的 。 ，本书会带 Spark 高 的 序

， 学 、 处理、图计算和 SQL 。 本书能 了解 Spark。

有 台电 有 大的集 ， Spark 能成 运 的数据分析工具。

本书的 读者 数据 学 和工 。 的 ， 能 用

Spark 解决 能会 有 法解决的问题。Spark 了 能丰 的数据

作 MLlib ， 数据 学 用 自 的 计学 ， 数据集

大 所能处理 的数据问题。 同时，工 本书中学 和 用

Spark 编 用的分 式 序 运 应用。工 和数据 学 能 本书中学

自需要的具 技能，而 能 自领 中 用 Spark 解决大 分 式问题。

数据 学 数据中发 及 。数据 学 有 计学 者数 学 ， 中的大 数 Python 、R 、SQL 等 数据分析工具。 本书 中， 会讲 Spark 中 学 和高 数据分析的 序 ， 会 Python

者 SQL 的应用作 Spark 用 进行 。 数据 学 ， 读

本书 ，能 快速 和 大数据 的同时， 用 的方式 解

决问题。

本书的第 读者 件工 。 工 ， 的 Java Python，

的编 ， 本书能 会 Spark 集 ， 用 Spark

shell， 及 编 Spark 应用 序 解决需要 行处理的问题。 Hadoop，

HDFS 进行交互 及 理集 的领 中领 了 。 有

Hadoop 用 心， 会 本书中讲解 本的分 式执行的 。

数据分析 工 ， 读 本书， 应 Python、Java、Scala 者

的编 有 本了解。 ， 有了 数据存 的解决方

，所 会讲 数据存 ， 会 的数据存

读 和保存数据。 用 编 心，有 的学

理解 ， 的 中列 了 。

本书 ， 次阅读的 序 的。 的开 ， 会

本 中的 数据 学 要，而 工 有用。

， 书中的所有 读者 能有 定的 。

会带 ， 自 的电 的 Spark， 用 Spark

能 有 本的 。等 了 Spark 的 和 Spark 的 ， 会

Spark shell。Spark shell 开发 Spark 应用 时 有用的工具。 会 Spark API、 Spark 应用运行 集 ， 及 Spark 所 的 高 的 序

， SQL 数据 和 MLlib 学 。

有 Python 的数据 学 ， Learning Python O’Reilly 和 Head First Python O’Reilly 。 有了 定的 Python ， Dive Into Python http://www.diveintopython.net/ 进 加 Python 的理解。

工 ，读 本书 高自 的数据分析技能，O’Reilly 出版的

Machine Learning for Hackers 和 Doing Data Science 的 书。

本书 要 学者而 ， 计 理解 Spark 理的人

本 加 的书。

本书 用了 列 版 定。

•

新术 。

• 等 字

序 ， 及 中出 的 量、 数 、数据 、数据 、 量、

和 字等。

• 加 等 字 Constant width bold

应 由用 的 本。

图 。

图 。

本书中所有的 代 GitHub 。 https://github.com/databricks/

learning-spark 中 和 出 代 。 代 Java、Scala 和 Python 的版本。

Java 版本的 代 Java 6 及 高版本。Java 8 了 lambda 的

新 法， 方 地编 数而简化Spark 代 。由

有开 用Java 8， 决定 的大 数 中 用 新 法。

Java 8 的 法 ， 阅Databricks Java 8 法的

http://databricks.com/blog/2014/04/14/spark-with-java-8.html 。有

序 会 Java 8 ， 发 本书的GitHub 代 中。

本书 要 成工作的。 ， 本书 了 代 ， 它用 的

序 中。 用了 大 分代 ， 需 。 ，用本书

的 代 序 需 ， 分发 O’Reilly 图书的 需要

用本书中的 代 问题 需 ， 书中大量的代 的

中 需要 。

制要 用本书 时加 用 。 用 书 、

作 者、 出 版 社 和 ISBN。 ： Web Development with Node and Express by Ethan Brown (O’Reilly). Copyright 2014 Ethan Brown, 978-1-491-94930-6.

自 代 的用法 出了 的 ， permissions@

oreilly.com 。

Safari ^® Books Online

Safari Books Online http://www.safaribooksonline.com 应运 而生的数字图书 。它同时 图书和 的 式出版

技术和 作 的 作 。技术 、 件开发人员、Web

计 、 人 和 等， 开 、解决问题、学

和 时， Safari Books Online 作 的 道。

、 和 人，Safari Books Online 合 和 的 定 价 。 用 能 的 数 据 问 O’Reilly Media、Prentice Hall Professional、Addison-Wesley Professional、Microsoft Press、Sams、Que、Peachpit Press、Focal Press、Cisco Press、John Wiley & Sons、Syngress、Morgan Kaufmann、IBM Redbooks、Packt、Adobe Press、FT Press、Apress、Manning、New Riders、McGraw-Hill、

Jones & Bartlett、Course Technology 及 出版社的 千 图书、 和 式出版 的书 。要了解 Safari Books Online 的 ， 网 。

本书的 价和问题发 出版社。

美 ：

O’Reilly Media, Inc.

1005 Gravenstein Highway North

Sebastopol, CA 95472

中 ：

北京市 区 大 2 号成 大 C 807 100035

技术 北京 有

O’Reilly 的 本书 有 网 ， 本书的 ， 、

代 及 。本书的网 地址 ：

http://bit.ly/web_dev_node_express

本书的 和技术 问题， 发 电子邮件 ：bookquestions@oreilly.com。

要了解 O’Reilly 图书、 、会 和新 的 ， 问 网 ： http://www.oreilly.com

Facebook 的地址 ：http://facebook.com/oreilly 的 Twitter ：http://twitter.com/oreillymedia

的 YouTube 地址 ：http://www.youtube.com/oreillymedia

Joseph Bradley、Dave Bridgeland、Chaz Chandler、Mick Davies、Sam DeHority、Vida Ha、Andrew Gal、Michael Gregson、Jan Joeppen、Stephan Jou、Jeff Martinez、Josh Mahonin、Andrew Or、Mike Patterson、Josh Rosen、Bruce Szalwinski、Xiangrui Meng、

Reza Zadeh 等审阅者， 本书的 作 出了 的 。

特 David Andrzejewski、David Buttler、Juliet Hougland、Marek Kolodziej、Taka Shinagawa、

Deborah Siegel、Normen Müller 、 Ali Ghodsi、Sameer Farooqui 等人， 大 分

了 的 ， 出了 要的 进 处。

要 编辑和编 分 的 题 。第 10 Tathagata Das 的 合作 共同 成的。Tathagata 了 大的 ， 的工作

、 问、 进 版 及 技术的 。Michael Armbrust 审校了 Spark

SQL 。 第 11 中，Joseph Bradley MLlib 了 。Reza

Zadeh 的 分 了图 和代 。Xiangrui Meng、Joseph Bradley 和

Reza Zadeh MLlib 了编审和 技术 的 。

第 1 章

Spark数据分析导论

本 会 Spark 。 Spark 和 件有 定了解，

第 2 开 读。

1.1 Spark

Spark 用 快速而通用的集 计算的 台。

速 方 ，Spark 了 用的 MapReduce 计算 ，而 高效地 计算

式， 交互式 和 处理。 处理大 数据集时，速 要的。速 快

进行交互式的数据 作， 次 作 需要等 数分 数 时。

Spark 的 要特 能 存中进行计算， 而 快。

进行的 计算，Spark MapReduce 加高效。

的 ，Spark 适用 需要 同的分 式 台的 ， 处理、

迭代算法、交互式 、 处理。 的 同的计算， Spark

简 而 地 处理 整合 。而 的 合， 的数据分析

中 有 义的。 ， Spark 的 特 大大 了 需要 台分

理的 。

Spark 所 的 丰 。 了 Python、Java、Scala 和 SQL 的简 用的

API 及 的丰 的 序 ，Spark 能和 大数据工具 合 用。 ，

Spark 运行 Hadoop 集 ， 问 Cassandra 的任 Hadoop 数据 。

1.2 大

Spark 集成的 件。Spark 的核心 由 计算任 成的、运行

工作 者 计算集群 的应用进行 、分发 及 的计算 。由

Spark 的核心 有 速 快和 用的特 ， Spark 同应用

计的高 件， SQL 和 学 等。 件 互 用，

件 中 用 序 ， 合 用 的 件。

件 合的 计 理有 。 ， 件 中所有的 序 和高 件

的 进中 。 ， Spark 的核心 新 了 化时，SQL 和

学 序 能自 能 。 次，运行 件 的代价 了。 需要运

行 5 10 的 件 了， 需要运行 件 。 代价

的 、 、 、 等。 Spark 件 中 增加 新的 件，

用 Spark 的 能 用新加 的 件。 新的数据分析 所需

要的 、 学 新的 件 的代价简化成了 需要 Spark。

， 合的 理的 大 ， 能 出 合 同处理 的应用。

， 用 Spark， 应用中 数据 中的数据 用 学 算法进行

时分 。 同时，数据分析 SQL 时 数据， 数据

化的 件进行 作。 ，有 的数据工 和数据 学

Python shell 问 数据，进行 时分析。 人 的 处理应用 问

数据。IT 需要 。

Spark 的 件 图 1-1 所 ， 次简要 它 。

MLib 学

GraghX 图计算 Spark SQL

化数据

Spark Streaming

时计算

图 1-1：Spark 软件栈

1.2.1 Spark Core

Spark Core 了 Spark 的 本 能， 任 、 存 理、 、 存

交互等 。Spark Core 中 了 弹性分布式数据集 resilient distributed dataset，简

RDD 的 API 定义。RDD 分 计算 行 作的元 集合，

Spark 要的编 。Spark Core 了 和 作 集合的 API。

1.2.2 Spark SQL

Spark SQL Spark 用 作 化数据的 序 。 Spark SQL， 用 SQL 者 Apache Hive 版本的 SQL 方 HQL 数据。Spark SQL 数据 ，

Hive 、Parquet 及 JSON 等。 了 Spark 了 SQL ，Spark SQL 开发者 SQL 和 的 RDD 编 的数据 作方式 合， 用 Python、Java

Scala，开发者 的应用中同时 用 SQL 和 的数据分析。 Spark

所 的丰 的计算 进行 的 合，Spark SQL 开 数据 工具

中 而出。Spark SQL Spark 1.0 中 的。

Spark SQL ，加 大学 分校 Apache Hive 运行 Spark

， 时的 作 Shark。 ，由 Spark SQL Spark 和 API 的 合 ， Shark Spark SQL 所 代。

1.2.3 Spark Streaming

Spark Streaming Spark 的 时数据进行 式计算的 件。 生 中的网

， 网络 中用 交的 新 成的 列， 数据 。 Spark

Streaming 了用 作数据 的 API， Spark Core 中的 RDD API 高 应。

， 序员编 应用时的学 ， 作 存 中的数据，

作 时数据 ， 序员 能应 自 。 计 ，Spark Streaming Spark Core 同 的 、 量 及 。

1.2.4 MLlib

Spark 中 的 学 ML 能的 序 ， 作 MLlib。MLlib

了 学 算法， 分 、 、 、 同 等， 了 、数据

等 的 能。MLlib 了 的 学 ， 用的

化算法。所有 方法 计 集 的 。

1.2.5 GraphX

GraphX 用 作图 社交网络的 图 的 序 ， 进行 行的图计算。

Spark Streaming 和 Spark SQL ，GraphX 了 Spark 的 RDD API，能用

和 任 的有 图。GraphX 图的 作 进行图

分 的

和 作所有 的

， 及 用图算法 PageRank 和 计数 。

1.2.6

而 ，Spark 计 高效地 计算 数千 计算 计

算。 了 的要 ，同时 大 ，Spark 集群管理器 cluster

manager 运行， Hadoop YARN、Apache Mesos， 及 Spark 自带的 简

， 作 。 要 有 任 集 理 的 Spark， Spark

自带的 而 有了 有 Hadoop YARN Mesos

的集 ， Spark 集 理 的 ， 的应用 同 能运行 集 。第

7 会 同的 及 合适的集 理 。

1.3 Spark

Spark 用 集 计算的 用计算 ， 用 的应用 序。 中

了本书的 大 读者人 ：数据 学 和工 。 分析 及

用 Spark 的方式， 发 用 Spark 的 用 ，

用 大 分 ——数据科学应用和数据处理应用。

， 领 和 用 式的 分 的。 人 有数据 学 和工 的能

，有的时 数据 学 的 进行 ， 成 工 ， 地编

的数据处理 序。 ，分开 大 和 应的用 有 义的。

1.3.1 数据

数据 学 年 出 的新学 ， 的 数据分析领 。 有 的定义，

数据科学家 data scientist 要 责分析数据 的人。数据 学 有

能具 SQL、 计、 学 等方 的 ， 及 定的 用 Python、

Matlab R 进行编 的能 。 数据 方 分析和 的 式， 数

据转换 data wrangling ，数据 学 中的 要技术有所了解。

数据 学 用 的技能 分析数据， 问题 发 。 的工作

会用 时分析，所 用交互式 shell 代 应用的 ，

时 和 简 代 的运行 。Spark 的速 及简 的 API 能

大 ，而 Spark 的 序 的 算法能 用。

Spark 列 件 数据 学任 。Spark shell Python 和 Scala 的

， 方 地进行交互式数据分析。Spark SQL 的 SQL shell，

shell 中 用 SQL 数据， 的 Spark 序 者 Spark shell 进 行 SQL 。 学 和数据分析 MLlib 序 。 ，Spark 能

用 R 者 Matlab 成的 序。数据 学 用 R Pandas 等 数据分析工具 时所能处理的数据集 ，而有了 Spark， 能处理 大数据 的问题。

的 ，数据 学 的工作需要 应用 中。具 问题 应用

的 能、 高应用的 定 ， 生 进行 ， 成 应用的 分。

， 数据 学 成 的 ， 能 会 生 中的

， 合 网 应用中， 用 。 ， 数据 学 的工作 化

生 中的应用的工作 由 的工 者工 成的，而 数据

学 。

1.3.2 数据

Spark 的 要用 工 的。 ， 工 定义 用 Spark 开发

生 中的数据处理应用的 件开发者。 开发者 有 本的 件工 ，

、 计 及 的编 ， 有计算 的 ， 能 用工

技术 计和 件 ， 用 。

工 ，Spark 开发用 集 行执行的 序 了 捷 。 ，Spark

需要开发者 分 式 编 的 问题， 需 网络 和

序 。Spark 工 了 的 快速 的任 ， 及 应用

进行 、审 和 能 。 API 化的特 分 式的 集 用

序 进行开发 及本地 大大简化。

Spark 用 所 Spark 开发 的数据处理应用， Spark 了丰 的

能， 学 和 用， 成 定。

1.4 Spark

Spark 由 大而 的开 社区开发和 的，社区中的开发者 自

同的 。 者 所 的 第 次 用 Spark， 会 Spark

的 。Spark 2009 年作 加 大学 分校 RAD

AMPLab 的 生。 中的 人员 用 Hadoop MapReduce。

发 MapReduce 迭代计算和交互计算的任 效 。 ，Spark 开

交互式 和迭代算法 计的，同时 存式存 和高效的 制。

2009 年， Spark 的 学术会 发 ，同年 Spark 式 生。 ，

MapReduce，Spark 任 了 10 20 的 能 。

Spark 的 分用 自加 分校的 ， 中 著 的有 Mobile

Millennium。作 学 领 的 ， 用 Spark 区

的交 。 了 的 时 ， 开 用 Spark。 ，

有 50 自 加 了 用 Spark 的 列 https://cwiki.apache.org/

confluence/display/SPARK/Powered+By+Spark 。 Spark 社区 的社区 Spark Meetups http://www.meetup.com/spark-users/ 和 Spark 会 http://spark-summit.org/ 中，

大 分 特有的 Spark 应用 。 了加 大学 分校，

Spark 作出 的 要 有 Databricks、 及英特尔。

2011 年，AMPLab 开 Spark 开发 高 的 件， Shark Spark 的 Hive

和

Spark Streaming。 件和 件 数据分析工具 BDAS，

https://amplab.cs.berkeley.edu/software/ 。

Spark 2010 年 3 月开 ， 2013 年 6 月交 了 Apache 会， 成了 Apache 开 会的 。

1.5 Spark

自 出 ，Spark 的 ，Spark 社区 保 的

。 版本号的 迭，Spark 的 者 增。Spark 1.0 了 100 开

序员 开发。 速地 ，Spark 社区 保 的发 新

版本的 。2014 年 5 月，Spark 1.0 式发 ，而本书 要 Spark 1.1.0 及

的版本。 ，大 数 版本的 Spark 中 适用，而大 数 能运行 版

本的 Spark 。

1.6 Spark

Spark 任 Hadoop 分 式 件 HDFS 的 件读 分 式数据集，

Hadoop 的 ， 本地 件、 S3、Cassandra、Hive、

HBase 等。 需要 的 ，Hadoop Spark 的 要 件，Spark 任

了 Hadoop 的存 。Spark 的 Hadoop 式 本 件、SequenceFile、

Avro、Parquet 等。 会 第 5 读 和存 时 数据 进行交互。

1：Shark Spark SQL 所 代。

第 2 章

Spark下载与入门

本 中， 会 Spark 本地 式 运行它。本 Spark 的所有 学

者的， 数据 学 和工 读。

Spark Python、Java Scala 用

。要用 本书 需要高 的编 技 ，

需要 中 的 法有 本的了解。 会 能 中 出 的

代 。

Spark 本 用 Scala 的，运行 Java JVM 。要 的电 集 运行 Spark， 要 的 工作 Java 6 者 新的版本。 用 Python ，

需要 Python 解 2.6 版本 。Spark Python 3

。

2.1 Spark

用 Spark 的第 和解 。 编译版本的 Spark 开 。 问 http://

spark.apache.org/downloads.html， Pre-built for Hadoop 2.4 and later Hadoop 2.4 及 新版本 编译的版本 ， Direct Download 。

的 TAR 件， 件 spark-1.2.0-bin-hadoop2.4.tgz.

1：Spark 1.4.0 加了R 。

2：Spark 1.4.0 Python 3。——译者

Windows 用 Spark 带有 的路 ， 能会 问

题。所 需要 Spark 带 的路 ， C:\spark 的

中。

需要 Hadoop， 有了 Hadoop 集 的 HDFS，

应版本的 Spark。 http://spark.apache.org/downloads.html 所需要的

， 会 的 件 有 同。 代 编译。

GitHub 新代 ， Source Code 代 进

行 。

大 数 Unix ， OSX 和 Linux， 有 tar的 行工具，

用 解 TAR 件。 的 作 有 tar， 网

络 的TAR 解 工具。 ， 用的 Windows，

7-Zip.

了 Spark ， 要进行解 ， 的 Spark 发行版中 有 。

打开 ， 工作路 的 Spark 所 的 ， 解开 。 会

出 和 同 了 .tgz 的新 件 。 工作路

新 有 。 用 成：

cd ~

tar -xf spark-1.2.0-bin-hadoop2.4.tgz cd spark-1.2.0-bin-hadoop2.4

ls

tar

所 的 行中，

记 定

执行解 作，

记 定 的

件 。

列出了 Spark 中的 。 地 Spark 中的

要的 件及 的 字和作用。

• README.md

用 Spark 的简 的 用 。

• bin

用 和 Spark 进行 方式的交互的 列 执行 件， 本 会讲 的 Spark shell。

• core、streaming、python……

• Spark 要 件的 代 。

• examples

和运行的 Spark 序， 学 Spark 的 API 有 。

要 Spark 数量 大的 件和 的 ， 会 本书 的 分中

讲解它 中的 大 分。 ， ， Spark 的 Python 和

Scala 版本的 shell。 运行 Spark 自带的 代 开 ， 编 、编译 运 行 自 简 的 Spark 序。

本 所 的 ，Spark 本地 式 运行， 分 式 式，

需要用 台 。Spark 运行 式 ， 了本地 式， 运行

Mesos YARN ， 运行 Spark 发行版自带的 。 会 第 7

讲 式。

2.2 Spark中Python Scala shell

Spark 带有交互式的 shell， 作 时数据分析。 用 R、Python、Scala 所 的 shell， 作 的 shell Bash 者 Windows 中的 ， 会 Spark shell 。 而和 shell 工具 的 ， shell 工具中 能 用 的 和 存 作数据，而 Spark shell 用 分 式存 的 存 者 的数据进行交互， 处理 的分发由 Spark 自 制 成。

由 Spark 能 工作 数据读 存中，所 分 式计算

成， 处理 TB 的数据的计算。 需要

shell 中 成的 交互式的 时 分析 适合 Spark。Spark Python 及 Scala 的增 版 shell， 集 的 。

本书中大 数 代 Spark 的所有 版本， 交互式shell

分 了Python 和 Scala 版本的 。shell 学 API 有

的， 读者 Python 和 Scala 版本的 子中 进行 ，

Java 开发者 ， 的API 的。

Spark shell 的 大 处 简 的方法 用 的 shell 作 简 的数据分 析。 Spark 方 中的快速 http://spark.apache.org/docs/latest/

quick-start.html 中的 。

第 打开 Spark shell。要打开 Python 版本的 Spark shell， 所 的 PySpark Shell，进 的 Spark ：

bin/pyspark

Windows 中 运行

。 要打开 Scala 版本的 shell， ：

等数 ，shell 会出 。Shell 时， 会 出。有的时 ，

由 出了 ， 需要 ， 的 shell 。

图 2-1 PySpark shell 时的 子。

图 2-1：默认日志选项下的 PySpark shell

shell 中 出的 而 人分心， 的 制 出的

量。 需要 conf log4j.properties 的 件 理 。Spark 开发者 Spark 中加 了 件的 版， 作 log4j.properties.template。

要 ， 版 件 制 conf/log4j.

properties 作 件， 行：

log4j.rootCategory=INFO, console

的 定 ， 及 的 ：

log4j.rootCategory=WARN, console

时 打开 shell， 会 出大大 图 2-2 。

图 2-2：降低日志级别后的 PySpark shell

IPython

IPython Python 用者 的增 版Python shell，能 自

等 用的 能。 http://ipython.org 。 要

量IPYTHON的 1， 用IPython 了：

IPYTHON=1 ./bin/pyspark

要 用IPython Notebook， Web 版的 IPython， 运行：

IPYTHON_OPTS="notebook" ./bin/pyspark

Windows ， 量 运行 行：

set IPYTHON=1 bin\pyspark

Spark 中， 分 式数据集的 作 的计算 图， 计算会自 地

集 行进行。 的数据集 弹性分布式数据集 resilient distributed dataset ， 简 RDD。RDD Spark 分 式数据和计算的 本 。

地 RDD ， 用 shell 本地 本 件 RDD 作

简 的 时 计。 2-1 Python 版的 子， 2-2 Scala 版的。

2-1：Python 行数 计

>>> lines = sc.textFile("README.md") # lines的RDD

>>> lines.count() # 计RDD中的元 数 127

>>> lines.first() # RDD中的第 元 ， README.md的第 行 u'# Apache Spark'

2-2：Scala 行数 计

scala> val lines = sc.textFile("README.md") // lines的RDD lines: spark.RDD[String] = MappedRDD[...]

scala> lines.count() // 计RDD中的元 数 res0: Long = 127

scala> lines.first() // RDD中的第 元 ， README.md的第 行 res1: String = # Apache Spark

要 出任 shell， Ctrl-D。

能 的 出 中 了 行 ：INFO SparkUI: Started

SparkUI at http://[ipaddress]:4040。 由 地址 问Spark 用

， 任 和集 的 。 会 第7 中 。

2-1 和 2-2 中， 量

RDD， 电 的 本地的 本 件

出 的。 RDD 运行 行 作， 计 数据集中的元 数

本的行数 ， 者 出第 元 。 会 中 RDD。

， 时 了解 Spark 的 本 。

2.3 Spark 心

用 shell 运行了 的第 Spark 序， 时 Spark 编 作 的了解了。

， Spark 应用 由 驱动器程序 driver program 发 集 的 行 作。 序 应用的

数， 定义了集 的分 式数据集，

分 式数据集应用了 作。 的 子 ， 的 序 Spark shell 本

， 需要 要运行的 作 了。

序

问 Spark。 代 计算集 的

。shell 时 自 了

， 作

的 量。

2-3 中的方法 出

它的 。

2-3： 量

>>> sc

<pyspark.context.SparkContext object at 0x1025b8f90>

有了 SparkContext， 用它 RDD。 2-1 和 2-2 中， 用了

sc.textFile()

代 件中 行 本的 RDD。 行 进行

作，

。

要执行 作， 序 要 理 执行器 executor 。 ，

集 运行

作， 同的 会 计 件的 同 分的行数。由

本地 式 运行 Spark shell， 所有的工作会 执行， shell 集 进行 行的数据分析。图 2-3 了 Spark 集 运行。

序

工作

工作 执行 执行

任

任 任

任

图 2-3：Spark 分布式执行涉及的组件

， 有 用 数的 API， 应 作运行 集 。 ，

的 README ， 出 件中 特定 的行。 Python

，具 代 2-4 Python 版本 和 2-5 Scala 版本 所 。 2-4：Python 版本 的 子

>>> lines = sc.textFile("README.md")

>>> pythonLines = lines.filter(lambda line: "Python" in line)

>>> pythonLines.first() u'## Interactive Python Shell'

2-5：Scala 版本 的 子

scala> val lines = sc.textFile("README.md") // lines的RDD lines: spark.RDD[String] = MappedRDD[...]

scala> val pythonLines = lines.filter(line => line.contains("Python"))

pythonLines: spark.RDD[String] = FilteredRDD[...]

scala> pythonLines.first()

res0: String = ## Interactive Python Shell

Spark 数

如果你对例 2-4 和例 2-5 中的

或者

语法不熟悉，可以把它们理解为 Python 和 Scala 中定义内联函数的简写方法。当你在这些语言中使用 Spark 时，你也可以单独 定义一个函数，然后把函数名传给 Spark。比如，在 Python 中可以这样做：

def hasPython(line):

return "Python" in line

pythonLines = lines.filter(hasPython)

在 Java 中向 Spark 传递函数也是可行的，但是在这种情况下，我们必须把函数定义为

实现了

接口的类。例如：

JavaRDD<String> pythonLines = lines.filter(

new Function<String, Boolean>() {

Boolean call(String line) { return line.contains("Python"); } }

);

Java 8 提供了类似 Python 和 Scala 的 lambda 简写语法。下面就是一个使用这种语法的 代码的例子：

JavaRDD<String> pythonLines = lines.filter(line -> line.contains("Python"));

我们会在 3.4 节更深入地讨论如何向 Spark 传递函数。

会 地讲 Spark API， ， Spark API 的地方

filter

数的 作 会 集 行执行。 ，Spark 会自

数

发 执行 。 ， 的

序中编 ， 代 自 运行 。第 3 会 讲 RDD API。

2.4

的 Spark 中的 分 序中 用 Spark。 了交互式运行

，Spark Java、Scala Python 的 序中 用。 shell 中 用 的 要区 需要自行 化 SparkContext。 ， 用的 API 了。

Spark 的 中 。 Java 和 Scala 中， 需要 的应用 加

spark-core

工件的 Maven 。编 书时，Spark 的 新版本 1.2.0， 应的

Maven ：

groupId = org.apache.spark artifactId = spark-core_2.10 version = 1.2.0

Maven 行的 理工具， 用 任 Java 的 ， 共

中的 序 。 用 Maven 的工 ， 用 能 问 Maven 的

工具 进行 ， Scala 的 sbt 工具 者 Gradle 工具。 用的集成开发

Eclipse Maven 加 工 中。

Python 中， 应 用 成 Python 本， 需 要 用 Spark 自带的

submit

本 运行。

本会 Python 序的 Spark 。 本

Spark 的 PythonAPI 了运行 。 需要 2-6 所 的 运行 本 。 2-6：运行 Python 本

bin/spark-submit my_script.py

， Windows 需要 用 代 。

2.4.1 SparkContext

成了应用 Spark 的 ， 需要 的 序中 Spark

SparkContext。

的应用，

SparkConf SparkContext 。 2-7 2-9 中， 用 分 了

。

2-7： Python 中 化 Spark

from pyspark import SparkConf, SparkContext

conf = SparkConf().setMaster("local").setAppName("My App") sc = SparkContext(conf = conf)

2-8： Scala 中 化 Spark

val conf = new SparkConf().setMaster("local").setAppName("My App") val sc = new SparkContext(conf)

2-9： Java 中 化 Spark

import org.apache.spark.SparkConf;

import org.apache.spark.api.java.JavaSparkContext;

SparkConf conf = new SparkConf().setMaster("local").setAppName("My App");

JavaSparkContext sc = new JavaSparkContext(conf);

子 了 SparkContext 的 本的方法， 需 数：

• 集群 URL： Spark 集 。 子中 用的

，

特 Spark 运行 而 需 集 。

• 应用名： 子中 用的

。 集 时，

集 理 的用 中 的应用。

有 加 数 用 应用的运行方式 加要发 集 的代 。 会

本书的 中 。

化 SparkContext ， 用 的所有方法 用 本 件

RDD 它 。

， Spark 用 SparkContext 的

方法， 者 出应用

者

。

快速 应 电 运行 的 Spark 应用了。 要了解 高

的 ，第 7 会讲 的应用 集 ， 的应用打 ，

代 自 发 工作 。 而 ， Spark 方 的快速

http://spark.apache.org/docs/latest/quick-start.html 了。

2.4.2

作 本讲大数据的书， 有 数 计的 子， 能成 的 。

数 计 ， 分 式 ，由 要 工作 读 合

数据， 数 计 成了 用的 子。 学 用 sbt 及 Maven

打 简 的 数 计的 。 所有的 ， 了

简 的 ， 保 了 本的 。 learning-spark-examples/mini-complete- example ， 的 工 。Java 版本 2-10 和 Scala 版本

2-11 的 子分 所 。

2-10：Java 版本的 数 计应用 时 需要

SparkConf conf = new SparkConf().setAppName("wordCount");

JavaSparkContext sc = new JavaSparkContext(conf);

// 读 的 数据

JavaRDD<String> input = sc.textFile(inputFile);

// 分

JavaRDD<String> words = input.flatMap(

new FlatMapFunction<String, String>() { public Iterable<String> call(String x) { return Arrays.asList(x.split(" "));

}});

// 计数

JavaPairRDD<String, Integer> counts = words.mapToPair(

new PairFunction<String, String, Integer>(){

public Tuple2<String, Integer> call(String x){

return new Tuple2(x, 1);

}}).reduceByKey(new Function2<Integer, Integer, Integer>(){

public Integer call(Integer x, Integer y){ return x + y;}});

// 计出 的 数存 本 件， 发

counts.saveAsTextFile(outputFile);

2-11：Scala 版本的 数 计应用 时 需要

val conf = new SparkConf().setAppName("wordCount") val sc = new SparkContext(conf)

// 读 的 数据

val input = sc.textFile(inputFile)

// 它 分成

val words = input.flatMap(line => line.split(" "))

// 计数

val counts = words.map(word => (word, 1)).reduceByKey{case (x, y) => x + y}

// 计出 的 数存 本 件， 发

counts.saveAsTextFile(outputFile)

用 简 的 sbt 2-12 Maven 2-13 件 应用。

由 Spark Core 工作 的 classpath 中了，所 Spark Core 的

记

， 用 assembly 方式打 应用时， 会

打 assembly 中。

2-12：sbt 件

name := "learning-spark-mini-example"

version := "0.0.1"

scalaVersion := "2.10.4"

// 加 序

libraryDependencies ++= Seq(

"org.apache.spark" %% "spark-core" % "1.2.0" % "provided"

)

2-13：Maven 件

<project>

<groupId>com.oreilly.learningsparkexamples.mini</groupId>

<artifactId>learning-spark-mini-example</artifactId>

<modelVersion>4.0.0</modelVersion>

<name>example</name>

<packaging>jar</packaging>

<version>0.0.1</version>

<dependencies>

<dependency> <!-- Spark -->

<groupId>org.apache.spark</groupId>

<artifactId>spark-core_2.10</artifactId>

<version>1.2.0</version>

<scope>provided</scope>

</dependency>

</dependencies>

<properties>

<java.version>1.6</java.version>

</properties>

<build>

<pluginManagement>

<plugins>

<plugin> <groupId>org.apache.maven.plugins</groupId>

<artifactId>maven-compiler-plugin</artifactId>

<version>3.1</version>

<configuration>

<source>${java.version}</source>

<target>${java.version}</target>

</configuration> </plugin> </plugin>

</plugins>

</pluginManagement>

</build>

</project>

spark-core 记 了provided， 了 制 assembly 方式打

应用时的行 。第7 中会 。

定了 方式， 打 用

本执行 的

应用了。

本 Spark 所要用 的 列 量。 mini-

complete-example 中， 用 Scala 2-14 者 Java 2-15 进行 。 2-14：Scala 运行

sbt clean package

$SPARK_HOME/bin/spark-submit \

--class com.oreilly.learningsparkexamples.mini.scala.WordCount \ ./target/... (as above) \

./README.md ./wordcounts

2-15：Maven 运行

mvn clean && mvn compile && mvn package

$SPARK_HOME/bin/spark-submit \

--class com.oreilly.learningsparkexamples.mini.java.WordCount \ ./target/learning-spark-mini-example-0.0.1.jar \

./README.md ./wordcounts

要了解 应用 序 Spark 的 子， Spark 方 中的快速

http://spark.apache.org/docs/latest/quick-start.html 。第 7 中 会 地讲解

打 Spark 应用。

2.5

本 中， 讲 了 的本地 式 运行 Spark， 及 Spark 的 用方式，

交互式方式和 应用进行 用。 简 了 Spark 编 的核心

： 序 SparkContext 和 列 RDD， 进行 行 作。

中， 会 加 地 作 RDD。

第 3 章

RDD编程

本 Spark 数据的核心 —— 分 式数据集 Resilient Distributed Dataset，简 RDD 。RDD 分 式的元 集合。 Spark 中， 数据的所有 作

RDD、 化 有 RDD 及 用 RDD 作进行 。而 ，Spark 会自 RDD 中的数据分发 集 ， 作 行化执行。

由 RDD Spark 的核心 ， 数据 学 和工 应 读 读本 。

读者 交互式 shell 2.2 中 。 ，本 中的 代

本书的 GitHub https://github.com/databricks/learning-spark 中 。

3.1 RDD

Spark 中的 RDD 的分 式 集合。 RDD 分 分区，

分区运行 集 中的 同 。RDD Python、Java、Scala 中任 的 ， 用 自定义的 。

用 用 方法 RDD：读 数据集， 序 分发

序中的 集合 list 和 set 。 本书 的 中 用

textFile()

读 本 件作 字 RDD 的 ， 3-1 所 。

3-1： Python 中 用

字 的 RDD

>>> lines = sc.textFile("README.md")

出 ，RDD 的 作： 转 化 操 作 transformation 和 行 动 操 作

action 。转化操作会由 RDD 生成 新的 RDD。 ， 据 数

据 的 化 作。 的 本 件 中， 用 生成 存

Python 的字 的新的 RDD， 3-2 所 。 3-2： 用 化 作

>>> pythonLines = lines.filter(lambda line: "Python" in line)

方 ，行动操作会 RDD 计算出 ， 序中，

存 存 HDFS 中。

用的 行 作，它

会 RDD 的第 元 ， 3-3 所 。 3-3： 用

行 作

>>> pythonLines.first() u'## Interactive Python Shell'

化 作和行 作的区 Spark 计算 RDD 的方式 同。 任 时 定 义新的 RDD， Spark 会惰性计算 RDD。它 有第 次 行 作中用

时， 会 计算。 开 能会 有 ， 大数据领

有道理的。 ， 3-2 和 3-3， 本 件定义了数据， 中

Python 的行 出 。 Spark 运行

时 件中

所有的行 读 存 ， 会 存 ，而 要 中的

数据。 ， Spark 了解了 的 化 作 ，它 计算 时 需

要的数据。 ， 行 作

中， Spark 需要 件 第

的行 ，而 需要读 件。

， ， Spark 的 RDD 会 次 它 进行行 作时 新计算。

行 作中 用同 RDD， 用

Spark RDD 存

。 Spark 数据 化 同的地方， 用的 会 3-6 中列出。

第 次 化的 RDD 计算 ，Spark 会 RDD 的 保存 存中 分区方式

存 集 中的 ， 的行 作中， 用 数据了。

RDD 存 而 存中。 进行 化 能 有 ，

大 数据集 有 义的： 会 用 RDD， 有 要 存

，Spark 数据 计算出 。

作中， 会 用

数据的 分读 存中，

分数据。 ， 次 README 件中 Python 的行进行计算，

出 3-4 所 的 本。

1： 任 时 能进行 算 RDD 的 。 保存RDD 数据的 台

时，Spark 用 特 算出 的分区， 用 的。

3-4： RDD 化 存中

>>> pythonLines.persist

>>> pythonLines.count() 2

>>> pythonLines.first() u'## Interactive Python Shell'

的 ， Spark 序 shell 会 方式工作。

(1) 数据 出 RDD。

(2) 用

的 化 作 RDD 进行 化， 定义新的 RDD。

(3) Spark 需要 用的中 RDD 执行

作。

(4) 用行 作

和

等 发 次 行计算，Spark 会 计算进行

化 执行。

cache() 用 存 用persist() 的。

会 解， Spark 中 的 RDD 作。

3.2 RDD

Spark 了 RDD 的方式：读 数据集， 及 序中 集合进

行 行化。

RDD 简 的方式 序中 有的集合 SparkContext 的

方法， 3-5 3-7 所 。 方式 学 Spark 时 有用，它 shell 中

快速 出自 的 RDD， RDD 进行 作。 ，需要 的 ， 了开发

和 时， 方式用 ， 方式需要 的 数据集 台

的 存中。

3-5：Python 中的

方法

lines = sc.parallelize(["pandas", "i like pandas"])

3-6：Scala 中的

方法

val lines = sc.parallelize(List("pandas", "i like pandas"))

3-7：Java 中的

方法

JavaRDD<String> lines = sc.parallelize(Arrays.asList("pandas", "i like pandas"));

用的方式 存 中读 数据 RDD。 数据集的读 会 第 5

。 ， 了用 本 件读 存 字 的 RDD 的方法

SparkContext.textFile()

，用法 3-8 3-10 所 。

3-8：Python 中的

方法

lines = sc.textFile("/path/to/README.md")

3-9：Scala 中的

方法

val lines = sc.textFile("/path/to/README.md")

3-10：Java 中的

方法

JavaRDD<String> lines = sc.textFile("/path/to/README.md");

3.3 RDD

，RDD 作：转化操作和行动操作。RDD 的 化 作 新的 RDD 的 作，

和

，而行 作 序

的 作，会 发 的计算，

和

。Spark

化 作和行 作的方式 ， 理解 进行的 作的 要的。

特定的 数 化 作 行 作 ， 它的

： 化 作 的 RDD，而行 作 的 的数据 。

3.3.1

RDD 的 化 作 新 RDD 的 作。 会 3.3.3 讲 ， 化出 的 RDD

的， 有 行 作中用 RDD 时 会 计算。 化 作 元

的， ， 化 作 次 会 作 RDD 中的 元 。 所有的

化 作 的。

子， 定 有 件 log.txt， 有 ， 出 中的 。

用 的 化 作

。 次， 会 用 Spark

的 的 API 分 3-11 3-13 。 3-11：用 Python

化 作

inputRDD = sc.textFile("log.txt")

errorsRDD = inputRDD.filter(lambda x: "error" in x)

3-12：用 Scala

化 作

val inputRDD = sc.textFile("log.txt")

val errorsRDD = inputRDD.filter(line => line.contains("error"))

3-13：用 Java

化 作

JavaRDD<String> inputRDD = sc.textFile("log.txt");

JavaRDD<String> errorsRDD = inputRDD.filter(

new Function<String, Boolean>() {

public Boolean call(String x) { return x.contains("error"); } }

});

，

作 会 有的

中的数据。 ， 作会 新

的 RDD。

的 序中 用， 中 的 。

，要

中 出所有 warning 的行。 ， 用

化 作

打印出 error warning 的行数。 中用 Python 作了 ，

union()

数的用法 所有 中 的。

3-14：用 Python 进行

化 作

errorsRDD = inputRDD.filter(lambda x: "error" in x) warningsRDD = inputRDD.filter(lambda x: "warning" in x) badLinesRDD = errorsRDD.union(warningsRDD)

union() filter()

的 同 它 作 RDD 而 。 化 作 作任

数量的 RDD。

要 3-14 中等价的 ， 的方法 出要 error 要

warning 的行， inputRDD进行 次 。

要 的 ， 化 作， 有的 RDD 中 生出新的 RDD，Spark 会 用谱系 图 lineage graph 记 同 RDD 的 。Spark 需要用 需

计算 RDD， 图 化的 RDD 分数据时 所 的数据。

图 3-1 了 3-14 中的 图。

图 3-1：日志分析过程中创建出的 RDD 谱系图

3.3.2

了 化 作 有的 RDD 出新的 RDD， 有时，

数据集进行 的计算。行 作 第 的 RDD 作，它 会 的

序， 者 存 中。由 行 作需要生成 的 出，它

会 制执行 用 的 RDD 的 化 作。

中用 的 的 子， 能 出

的 。

，需要 用 行 作 ：用

计数 ，用

收集

RDD 中的 元 ， 3-15 3-17 所 。 3-15： Python 中 用行 作 进行计数

print "Input had " + badLinesRDD.count() + " concerning lines"

print "Here are 10 examples:"

for line in badLinesRDD.take(10):

print line

3-16： Scala 中 用行 作 进行计数

println("Input had " + badLinesRDD.count() + " concerning lines") println("Here are 10 examples:")

badLinesRDD.take(10).foreach(println)

3-17： Java 中 用行 作 进行计数

System.out.println("Input had " + badLinesRDD.count() + " concerning lines") System.out.println("Here are 10 examples:")

for (String line: badLinesRDD.take(10)) { System.out.println(line);

}

子中， 序中 用

了 RDD 中的 量元 。 本地

元 ， 打印出 。RDD 有

数， 用

RDD 中的数据。 的 序 RDD 的 ， 本地处理

数据时， 用它。记 ， 有 的 数据集能 台 的 存中

时， 能 用

， ，

能用 大 数据集 。

大 数 ，RDD 能

收集 进 中， 它 大。

时， 要 数据 HDFS Amazon S3 的分 式的存 中。

用

、

， 者任 的 行 作 RDD 的

数据 自带的 式保存 。 会 第 5 讲解 出数据的 。

需要 的 ， 用 新的行 作时， RDD 会 开 计算。要

效的行 ，用 中 化， 会 3.6 中 。

3.3.3

，RDD 的 化 作 的。 用行 作 Spark 会

开 计算。 新用 能 有 处， 用 Haskell

等 数式 者 LINQ 的数据处理 的人 ，会有 。

RDD 用 化 作 用

时， 作 会 执行。

，Spark 会 记 所要 执行的 作的 。 应 RDD 作存

特定数据的数据集，而 RDD 作 化 作 出 的、记

计算数据的 列 。 数据读 RDD 的 作 同 的。 ， 用

sc.textFile()

时，数据 有读 进 ，而 要时 会读 。和 化 作 的 ，

读 数据的 作 有 能会 次执行。

化 作 的， 时 运行 行 作 制

Spark 执行 RDD 的 化 作， 用count()。 所 的 序

进行 分 的简 方法。

Spark 用 ， 作合 计算数据的 。

Hadoop MapReduce 的 中，开发者 大量时 作 合 ，

MapReduce 的 数。而 Spark 中， 出 的 能 用

简 的 作 的 能。 ，用 用 的 作 的 序，

作 理。

3.4 Spark 数

Spark 的大 分 化 作和 分行 作， 需要 用 的 数 计算。

的 要 中， Spark 数的方式 有区 。

3.4.1 Python

Python 中， 有 方式 数 Spark。 的 数时， 用

lambda 式 ， 3-2 和 3-18 所 。 了 lambda 式，

数 定义的 数。

3-18： Python 中 数

word = rdd.filter(lambda s: "error" in s) def containsError(s):

return "error" in s

word = rdd.filter(containsError)

数时需要 心的 ，Python 会 数所 的 序列化 出

。 的 的成员， 者 了 中 字 的 用时

self.field

，Spark 会 发 工作 ， 能 的 大

3-19 。有时， 的 Python 道 序列化 的 ， 会

的 序 。

3-19： 带字 用的 数

def __init__(self, query):

self.query = query def isMatch(self, s):

return self.query in s

def getMatchesFunctionReference(self, rdd):

# 问题： "self.isMatch"中 用了 self return rdd.filter(self.isMatch)

def getMatchesMemberReference(self, rdd):

# 问题： "self.query"中 用了 self return rdd.filter(lambda x: self.query in x)

代的方 ， 所需要的字 中 出 量中，

量， 3-20 所 。

3-20： 带字 用的 Python 数

...

def getMatchesNoReference(self, rdd):

# ： 需要的字 量中

query = self.query

return rdd.filter(lambda x: query in x)

3.4.2 Scala

Scala 中， 定义的 数、方法的 用 方法 Spark，

Scala 的 数式 API 。 要 ， 所 的 数及 用

的数据需要 序列化的 了 Java 的 Serializable 。 ， Python ，

的方法 者字 时，会 的 用。 Scala 中 ，

会 Python 用

出 用。 3-20 中 Python 执行

的 作， 需要的字 量中， 字 的

， 3-21 所 。 3-21：Scala 中的 数

class SearchFunctions(val query: String) { def isMatch(s: String): Boolean = { s.contains(query)

}

def getMatchesFunctionReference(rdd: RDD[String]): RDD[String] = {

// 问题："isMatch" "this.isMatch"， 要 "this"

rdd.map(isMatch) }

def getMatchesFieldReference(rdd: RDD[String]): RDD[String] = { // 问题："query" "this.query"， 要 "this"

rdd.map(x => x.split(query)) }

def getMatchesNoReference(rdd: RDD[String]): RDD[String] = {

// ： 需要的字 出 量中

val query_ = this.query rdd.map(x => x.split(query_)) }

}

Scala 中出 了

， 问题 了 序列

化的 中的 数 字 。记 ， 序列化 量 中的 数 的。

3.4.3 Java

Java 中， 数需要作 了 Spark 的

中的任

数 的 。 据 同的 ， 定义了 同的 。

本的 数 列 3-1 中，同时 了 的 数 ， 需要 特

的数据时 用， 3.5.2 中的 Java 。 表3-1：标准Java函数接口

Function<T, R> R call(T)

map()

filter()

Function2<T1, T2, R> R call(T1, T2)

aggregate()

和

fold()

FlatMapFunction<T, R> Iterable<R> call(T)

flatMap()

的 作中

的 数 定义 用 3-22 ， 具

3-23 。

3-22： Java 中 用 进行 数

RDD<String> errors = lines.filter(new Function<String, Boolean>() { public Boolean call(String x) { return x.contains("error"); } });

3-23： Java 中 用具 进行 数

class ContainsError implements Function<String, Boolean>() { public Boolean call(String x) { return x.contains("error"); } }

RDD<String> errors = lines.filter(new ContainsError());