链接：http://spark.apache.org/docs/latest/programming-guide.html

安装好Spark 后，自带了一些demo, 路径在Spark根目录/examples/src/main/python/

里面有些例子，例如统计字数的 wordcount.py

import sys  from operator import add     from pyspark import SparkContext     import sys  reload(sys)  sys.setdefaultencoding("utf-8")     if __name__ == "__main__":      if len(sys.argv) != 2:          print >> sys.stderr, "Usage: wordcount <file>"          exit(-1)      sc = SparkContext(appName="PythonWordCount")      lines = sc.textFile(sys.argv[1], 1)      counts = lines.flatMap(lambda x: x.split(' ')) \                    .map(lambda x: (x, 1)) \                    .reduceByKey(add)      output = counts.collect()      for (word, count) in output:          print "%s: %i" % (word, count)         sc.stop()

另外参考Spary 的 python api: http://spark.apache.org/docs/latest/api/python/pyspark.html

写了一个小demo，就是练习一下api的使用，做业务很方便。针对于大数据文件做统计分析的。比如几十兆上百兆的我们单机处理，上G的就放在hadoop 的 hdfs上。

下面是一个学生成绩单。四列字段：学生，以及三科成绩。其中学生有重复的（比如额外加分的情况，需要合并分析）。

yang    85  90  30  wang    20  60  50  zhang   90  90  90  li  100 54  0  yanf    0   0   0     yang 12 0 0

当然实际中数据要很多，比如很多列，而且几十万行甚至几百万行。这里是一个demo ,相当于在部署前测试。

在 Spark根目录/example/src/main/python/ 下新建一个 students.py :

#coding=utf-8     import sys  from operator import add  from pyspark import SparkContext     import sys  reload(sys)  sys.setdefaultencoding("utf-8")     def map_func(x):      s = x.split()      return (s[0],[int(s[1]),int(s[2]),int(s[3])])     def f(x):      return x      rank = sc.parallelize(range(0,sorted.count()))     def add(a,b):      return [a[r]+ b[r] for r in range(len(a))]     def _merge(a,b):      print '****'      return [a[r]+ b[r] for r in range(len(a))]     #the students who has one score is 100  def has100(x):      for y in x:          if(y==100):              return True      return False     def allIs0(x):      if(type(x) == list and sum(x) == 0):          return True      return False     def subMax(x,y):      m = [x[1][i] if(x[1][i] > y[1][i]) else y[1][i] for i in range(3)]      return('',m)     def sumAll(x,y):      return ('',[x[1][i]+y[1][i] for i in range(3)])        if __name__ == "__main__":      if len(sys.argv) != 2:          print >> sys.stderr, "Usage: students <file>"          exit(-1)      sc = SparkContext(appName="Students")      #加载学生文件，调用map将学生映射成keyValues.其中,key是学生，Value是学生成绩。map后的结果如('yang',(85,90,30))      # 之后调用 CombineByKey,将相同学生的成绩相加（合并）。然后调用cache, 将整个数据缓存，以便多次进行reduce而无需每次都重新生成。      lines = sc.textFile(sys.argv[1], 1).map(map_func).combineByKey(f,add,_merge).cache()      #print lines      count = lines.count()             # 获取学生中三科成绩有满分的，调用filter来实现      whohas100 = lines.filter(lambda x: filter(has100,x)).collect()      # 获取三科中所有成绩都是0的同学（缺考）      whoIs0 = lines.filter(lambda x: filter(allIs0,x)).collect()      # 获取每个学生的成绩总和      sumScore = lines.map(lambda x: (x[0],sum(x[1]))).collect()      # 获取三科中，单科最高分      subM = lines.reduce(subMax)      # 获取学生单科成绩的总和，求单科平均分用      sumA = lines.reduce(sumAll)      # 总分最高的学生      maxScore = max(sumScore,key = lambda x: x[1])      # 总分最低的学生      minScore = min(sumScore,key = lambda x: x[1])      # 所有学生三科成绩平均分      avgA = [x/count for x in sumA[1]]      # 根据总分进行排序（默认由小而大）      sorted = lines.sortBy(lambda x: sum(x[1]))      # 排序并附带序号      sortedWithRank = sorted.zipWithIndex().collect()      # 取出成绩最高的前三名同学，发奖！      first3 = sorted.takeOrdered(3,key = lambda x: -sum(x[1]))            #print '*'*50      print whohas100      print maxScore      print whoIs0      print subM      print avgA      print sorted.collect()      print sortedWithRank      print first3             #将结果汇总输出到文件      file = open('/home/yanggaofei/downloads/result.txt','w')      file.write('students num:'+`count`+ '\n')      file.write('who has a 100 scores:' + str(whohas100) + '\n')      file.write('who all is 0:' + str(whoIs0) + '\n')      file.write('the max score of each subject:' + str(subM) + '\n')      file.write('the avg score of each subject:' + str(avgA) + '\n')      file.write('sorted the students:' + str(sorted.collect()) + '\n')      file.write('sorted the students with the rank:' + str(sortedWithRank) + '\n')      file.write('the first 3 who will get the award:' + str(first3) + '\n')      file.close()

好了，运行：

students num:5  who has a 100 scores:[(u'li', [100, 54, 0])]  who all is 0:[(u'yanf', [0, 0, 0])]  the max score of each subject:('', [100, 90, 90])  the avg score of each subject:[61, 58, 34]  sorted the students:[(u'yanf', [0, 0, 0]), (u'wang', [20, 60, 50]), (u'li', [100, 54, 0]), (u'yang', [97, 90, 30]), (u'zhang', [90, 90, 90])]  sorted the students with the rank:[((u'yanf', [0, 0, 0]), 0), ((u'wang', [20, 60, 50]), 1), ((u'li', [100, 54, 0]), 2), ((u'yang', [97, 90, 30]), 3), ((u'zhang', [90, 90, 90]), 4)]  the first 3 who will get the award:[(u'zhang', [90, 90, 90]), (u'yang', [97, 90, 30]), (u'li', [100, 54, 0])]