如何用Python写spark

Python011

如何用Python写spark,第1张

1.RDD是PariRDD类型

def add1(line):

return line[0] + line[1]

def add2(x1,x2):

return x1 + x2

sc = SparkContext(appName="gridAnalyse")

rdd = sc.parallelize([1,2,3])

list1 = rdd.map(lambda line: (line,1)).map(lambda (x1,x2) : x1 + x2).collect() #只有一个参数,通过匹配来直接获取(赋值给里面对应位置的变量)

list1 = rdd.map(lambda line: (line,1)).map(lambda x1,x2 : x1 + x2).collect() #错误,相当于函数有两个参数

list2 = rdd.map(lambda line: (line,1)).map(lambda line : line[0] + line[1]).collect() #只有一个参数,参数是Tuple或List数据类型,再从集合的对应位置取出数据

list3 = rdd.map(lambda line: (line,1)).map(add1).collect() #传递函数,将Tuple或List类型数据传给形参

list4 = rdd.map(lambda line: (line,1)).map(add2).collect() #错误,因为输入只有一个,却有两个形参

当RDD是PairRDD时,map中可以写lambda表达式和传入一个函数。

a、写lambda表达式:

可以通过(x1,x2,x3)来匹配获取值;或者使用line获取集合,然后从集合中获取。

b、传入函数

根据spark具体的transaction OR action 操作来确定自定义函数参数的个数,此例子中只有一个参数,从形参(集合类型)中获取相应位置的数据。

在ipython中使用spark

说明:

spark 1.6.0

scala 2.10.5

spark安装路径是/usr/local/spark;已经在.bashrc中配置了SPARK_HOME环境变量。

方法一

/usr/local/Spark/bin/pyspark默认打开的是Python,而不是ipython。通过在pyspark文件中添加一行,来使用ipython打开。

cp pyspark ipyspark

vi ipyspark

# 在最前面添加

IPYTHON=1

# 启动

ipyspark

方法二:

通过为spark创建一个ipython 配置的方式实现。

# 为spark创建一个ipython 配置

ipython profile create spark

# 创建启动配置文件

cd ~/.config/ipython/profile_spark/startup

vi 00-pyspark-setup.py

在00-pyspark-setup.py中添加如下内容:

import os

import sys

# Configure the environment

if 'SPARK_HOME' not in os.environ:

os.environ['SPARK_HOME'] = '/srv/spark'

# Create a variable for our root path

SPARK_HOME = os.environ['SPARK_HOME']

# Add the PySpark/py4j to the Python Path

sys.path.insert(0, os.path.join(SPARK_HOME, "python", "pyspark"))

sys.path.insert(0, os.path.join(SPARK_HOME, "python", "lib", "py4j-0.9-src.zip"))

sys.path.insert(0, os.path.join(SPARK_HOME, "python"))

启动ipython

ipython –profile spark

测试程序

在ipython中输入一下命令,如果下面的程序执行完后输出一个数字,说明正确。

from pyspark import SparkContext

sc = SparkContext( 'local', 'pyspark')

def isprime(n):

"""

check if integer n is a prime

"""

# make sure n is a positive integer

n = abs(int(n))

# 0 and 1 are not primes

if n <2:

return False

# 2 is the only even prime number

if n == 2:

return True

# all other even numbers are not primes

if not n &1:

return False

# for all odd numbers

for x in range(3, int(n**0.5)+1, 2):

if n % x == 0:

return False

return True

# Create an RDD of numbers from 0 to 1,000,000

nums = sc.parallelize(xrange(1000000))

# Compute the number of primes in the RDD

print 逗Result: 地, nums.filter(isprime).count()

方法三

将上面的程序放入test.py文件,执行命令python test.py。发现错误。因为没有将pyspark路径加入PYTHONPATH环境变量。

在~/.bashrc或/etc/profile中添加如下内容:

# python can call pyspark directly

export PYTHONPATH=$SPARK_HOME/python:$SPARK_HOME/python/pyspark:$SPARK_HOME/python/lib/py4j-0.9-src.zip:$PYTHONPATH

执行如下命令:

# 使配置生效

source ~/.bashrc

# 测试程序

python test.py

前段时间使用了一下google的博客空间,感觉也很一般,所以现在把那里的几篇文章转过来。

执行python脚本只需要对python文件做如下操作即可:

在python文件里第一行加上#!

/usr/bin/python,即你的python解释器所在的目录。另外还有一种写法是#!

/usr/bin/env

python

编辑完成python脚本文件后为它加上可执行权限。例如你的python脚本文件叫做runit.py,那么就在shell中输入如下命令:chmod

+x

runit.py

之后直接在shell中输入./runit.py就可以执行你的python程序了。

当然这是在Linux下的操作,如果想在windows下直接执行Python程序,就需要使用py2exe工具将python源程序编译成exe文件了。