Spark 的 Python 编程

Spark 集群的安装在我的 CSDN 博客 里面有详细介绍，这里不再赘述。另外，spark 的原理介绍有时间单独开一篇博客来讲，这篇博客主要介绍 spark 的 python 编程。

Spark 环境启动

因为最近在用 Pittsburgh Supercomputing Center(PSC) 的计算资源在学习 spark，所以这里主要介绍 PSC 的 spark 环境启动方法。

• 利用 ssh 登录 bridges.psc.edu
• 使用interact命令进入计算节点
• 使用module load spark命令载入 spark 模块
• 使用pyspark命令进入 python 交互环境，这个时候就可以开始按照 python 语法进行编程了。

如果想要执行 python 脚本，有以下两种方法：

• 进入 pyspark 环境后，执行execfile("python_file.py")
• 不进入 pyspark 编译环境，执行spark-submit python_file.py

如果是后一种方式，在编写正式 spark python 语句之前，需要在头部加入以下语句：

from pyspark import SparkContext, SparkConf

conf = SparkConf()
sc = SparkContext(conf=conf)

Spark 常用 transformation 和 action

Transformation

常规 transformation

• map(func): 分别利用 func 对现有 RDD 的每个元素进行运算，生成新 RDD
• filter(func): 选择使得通过 func 运算为真的元素，生成新 RDD
• flatMap(func): 与 map 相比较，func 可以返回多个元素，最终 flatMap 的一个元素可以生成多个元素
• distinct(): 返回元素不重复的 RDD

Key-value transformation

• reduceByKey(func): 通过 key 对现有 RDD 进行 reduce，reduce 方法由 func 指定
• sortByKey(): 返回经过 key 排序后的 RDD

Action

常规 action

• collect(): 返回 RDD 的所有元素
• count(): 返回 RDD 所有元素的个数
• take(n): 返回 RDD 前 n 个元素

Key-value action

• countByKey(): 返回每个 key 元素的个数
• lookup(key): 返回指定 key 的所有元素

一个例子

下面以一个例子来说明 spark python 的使用方法。本例子要实现的任务是对一个包含 Shakespeare 所有著作的文件，统计出现次数最多的几个词。Spark python 实现代码如下：

# 读入 txt 文件，放到 rdd 中，其中 sc 代表 spark context
rdd = sc.textFile("Complete_Shakespeare.txt");
# 在读入的 rdd 中，一个元素代表一行文字，下面这条语句把行拆分成单词，使得 words_rdd 的每个元素代表一个单词，其中 lambda 是 python 中嵌入式函数定义方法，分号前面是输入参数，后面是函数实现方法
words_rdd = rdd.flatMap(lambda x: x.split());
# 把 words_rdd 转化成 key-value 的形式，以充分利用 key-value transformation 的功能
key_value_rdd = words_rdd.map(lambda x: (x,1));
# 在 key_value_rdd 中，key 为单词，value 为 1，在 word_counts_rdd 中，key 为单词，value 为该单词出现的次数
word_counts_rdd = key_value_rdd.reduceByKey(lambda x,y: (x+y));
# key 和 value 互换位置
flipped_rdd = word_counts_rdd.map(lambda x: (x[1],x[0]));
# 互换位置后就可以通过 key 进行排序了
result_rdd = flipped_rdd.sortByKey(False);
# 取出出现次数最多的前 n 个单词
result_rdd.take(n)

Reference