开发者社区> 陈国林> 正文
阿里云
为了无法计算的价值
打开APP
阿里云APP内打开

Spark-python-快速开始

简介: 1. 概览 这篇文章主要是关于Spark的快速熟悉和使用,我们使用Python和Spark的shell接口来操作Spark。 Spark shell使得我们可以很简单的学习Spark的Api,同时也是一个强大数据分析交互的工具。
+关注继续查看

1. 概览

这篇文章主要是关于Spark的快速熟悉和使用,我们使用Python和Spark的shell接口来操作Spark。 Spark shell使得我们可以很简单的学习Spark的Api,同时也是一个强大数据分析交互的工具。

2. Spark shell

我们使用Python版本的Spark工具pyspark,前提是Spark的安装路径已经加到环境变量PATH中,否则会报找不到命令 特别说明: >>>表示的所pyspark的命令

./bin/pyspark

Spark核心的抽象是弹性分布式数据集合,我们称为RDD(Resilient Distributed Dataset)。一个RDD可以从输入文件中产生比如HDFS文件,也可以从其他RDD转换而来。

我们通过读取本地文件text.dat来创建一个新的RDD

hadoop@ubuntu:~/github$ cat text.dat
spark
i
love
you
>>> textFile = sc.textFile("text.dat")

RDD包括两种运算操作,action和transformation。action操作会返回值,例如count(),transformation操作则是返回一个新的RDD,例如filter()。

2). RDD action操作

>>> textFile.count()
4
>>> textFile.first()
u'spark'

3). RDD transformation操作

>>> newTextFile = textFile.filter(lambda line: "spark" in line)
>>> newTextFile.count()
1

4). RDD的操作支持链接在一起操作

>>> textFile.filter(lambda line: "spark" in line).count()
1

3. RDD更多操作

1). RDD的action和transformation可以用在更复杂的计算上面

>>> textFile.map(lambda line: len(line)%2).reduce(lambda a,b: a if (a > b) else b)
1

说明: map产生一个新的RDD,RDD每个值是一个整数,等于每一行长度的1/2.reduce产生另一个新的RDD,对于key相同的数据取整数值最大的那个.(默认情况下key都相同)

在这里我们采用的是python的lambda来代替函数, 所以上面的代码等价于下面的代码

>>> def max(a, b):
...     if a > b:
...             return a
...     else:
...             return b
... 
>>> textFile.map(lambda line: len(line)%2).reduce(max)
1

对于Hadoop的MapReduce来说,Spark可以很容易就实现,比如常见的word count

>>> wordCount = textFile.flatMap(lambda line: line.split('u')).map(lambda word: (word, 1)).reduceByKey(lambda a, b: a+b)
>>> wordCount.collect()
[(u'i', 1), (u'', 1), (u'spark', 1), (u'love', 1), (u'yo', 1)]

在这里我用到了flatMap 这个api, 把一行映射为多行并产生一个新的RDD.再用map把映射为pair(word, 1), 最后用reduce把相同word的count相加. 我们可以用collect action以list方式输出RDD数据

4. 数据缓存

Spark支持把数据缓存到内存中, 很多时候当某些数据被频繁利用的时候缓存数据到内存是非常有用的. 比如当我们在计算网页PageRank的时候可以把经常用的query集合缓存到内存中

>>> wordCount.cache()
PythonRDD[30] at collect at <stdin>:1
>>> wordCount.count()
5
>>> wordCount.count()
5

5. Spark应用程序

同理我们也可以使用Spark API编写Spark应用程序

现在我们用Python API编写一个Spark的应用程序, 命名为simple_app.py

"""
Simple spark app
"""

from pyspark import SparkContext
sc = SparkContext("local", "SimpleApp")

#Spark default read from HDFS
#must be sure has exist HDFS file /user/hadoop/test_data/README.md
data_rdd = sc.textFile('/user/hadoop/test_data/README.md').cache()
num_a = data_rdd.filter(lambda line: 'a' in line).count()
num_b = data_rdd.filter(lambda line: 'b' in line).count()
print num_a
print num_b

我们用spark-submit来提交这个应用程序

spark-submit --master local simple_app.py

快速的完成了Spark的第一个应用程序,后面我们会对Spark的各个模块进入更深的研究


版权声明:本文内容由阿里云实名注册用户自发贡献,版权归原作者所有,阿里云开发者社区不拥有其著作权,亦不承担相应法律责任。具体规则请查看《阿里云开发者社区用户服务协议》和《阿里云开发者社区知识产权保护指引》。如果您发现本社区中有涉嫌抄袭的内容,填写侵权投诉表单进行举报,一经查实,本社区将立刻删除涉嫌侵权内容。

相关文章
spark中 map和reduce理解及与hadoop的map、reduce区别
spark中 map和reduce理解及与hadoop的map、reduce区别
0 0
HADOOP MapReduce 处理 Spark 抽取的 Hive 数据【解决方案一】
开端: 今天咱先说问题,经过几天测试题的练习,我们有从某题库中找到了新题型,并且成功把我们干趴下,昨天今天就干了一件事,站起来。 沙问题? java mapeduce 清洗 hive 中的数据 ,清晰之后将driver代码 进行截图提交。
0 0
Storm与Spark、Hadoop三种框架对比
Storm与Spark、Hadoop这三种框架,各有各的优点,每个框架都有自己的最佳应用场景。所以,在不同的应用场景下,应该选择不同的框架。
0 0
阿里巴巴资深架构师熬几个通宵肛出来的Spark+Hadoop+中台实战pdf
阿里巴巴资深架构师熬几个通宵肛出来的Spark+Hadoop+中台实战pdf
0 0
大数据Spark企业级实战与Hadoop实战&PDF和PPT
今天给大家分享的是《大数据Spark企业级实战》与《Hadoop实战》《大数据处理系统·Hadoop源代码情景分析》《50个大厂大数据算法教程》等销量排行前10名的大数据技术书籍(文末领取PDF版)。这些书籍具有以下几个优点:易读、实践性强,对解决工作中遇到的业务问题具有一定启发性。
0 0
Spark SQL CLI部署CentOS分布式集群Hadoop上方法
Spark SQL CLI部署CentOS分布式集群Hadoop上方法
0 0
Apache Spark vs.Apache Hadoop
Apache Spark vs.Apache Hadoop
0 0
【Spark】Spark 与 Hadoop MR 之间的区别
【Spark】Spark 与 Hadoop MR 之间的区别
0 0
spark 3.1.x支持(兼容)hive 1.2.x以及hadoop cdh版本的尝试
spark 3.1.x支持(兼容)hive 1.2.x以及hadoop cdh版本的尝试
0 0
+关注
陈国林
曾任职于阿里巴巴,现就职于美图,专业搬砖100年~
文章
问答
文章排行榜
最热
最新
相关电子书
更多
Improving Python and Spark
立即下载
Improving Python and Spark Per
立即下载
From Python Scikit learn to Scala Spark
立即下载