运行Spark脚本前,启动Hadoop的HDFS和YARN。Spark的shell提供
了简单方式去识别API,同样也有一个强大的工具去交互式地分析数据。两种语言有这样的交换能力,分别是Scala 和 Python。下面将演示如何使用Python来分析数据文件。
进入Spark安装主目录,输入下面的命令,python命令行模式将会启动。
./bin/pyspark
Spark的主要抽象是一个称之为弹性分布式数据集(Resilient Distributed Dataset,RDD)。RDD能够从Hadoop输入格式(InputFormat,比如HDFS文件),或者通过转换其它RDDs。这里,我们将上传Spark主目录下的README文件上传到Hadoop的文件系统/user/app(app是我的linux用户名)目录下。具体的命令如下:
hdfs dfs –mkdir–p /user/app hdfs dfs –putREADME.md /usr/app
使用Python创建一个新的弹性分布式数据集,定义如下。
>>>textFile = sc.textFile("README.md")
RDD有actions,,它能够返回值以及转换transformations,也可以返回指向新的RDD的指针。下面就是几个RDD的action操作。
>>>textFile.count() # 返回这个RDD的数据项个数 126 >>>textFile.first() # 返回这个RDD的第一项数据 u'# ApacheSpark'
现在让我们使用一个转换(transformation),我们将使用filter转换返回一个新的RDD,伴随着文件中的数据项的子集。
>>>linesWithSpark=textFile.filter(lambdaline:"Spark"inline)
我们能够链式化转换和动作:
>>> textFile.filter(lambda line:"Spark"inline).count()# How many linescontain "Spark"? 15
RDD动作和转换能够被用作更为复杂的计算,看下面的例子。
>>> textFile.map(lambdaline: len(line.split())).reduce(lambda a, b: aif (a > b)elseb) 15
这个参数的第一项映射一行的数据到一个integer值,创建一个新的RDD。reduce在RDD上被调用,用于找到一个最大的行数。map和reduce的参数是Python匿名函数lambdas。与此同时,你也能够传入任何你想要的顶级Python函数。不如,下面就是传入的一个Python函数。
>>> defmax(a, b): ... if a > b: ... return a ... else: ... return b ... >>>textFile.map(lambda line: len(line.split())).reduce(max) 15
Spark快速上手之交互式分析,布布扣,bubuko.com
原文地址:http://blog.csdn.net/john_f_lau/article/details/38010287