基本数据结构RDD

弹性分布式数据集。既是数据集，又是编程模型（就是python里的对象）。

• 特点
  • RDD是数据集
  • RDD是编程模型，可以调用函数
  • RDD相互之间有依赖关系。（RDD.xxx()可以生成新的RDD）
  • RDD是可以分区的。RDD的一些方法可以访问HDFS，HDFS是分区的。RDD可以访问很多种数据库。RDD处理非常大的数据集，所以支持分区，是并行的计算框架

pyspark使用方法

pyspark 手册

常用API

SparkContext()

• 入口API，能够创建RDD，设置参数和Jar包

mapPartitions(f, numSlice，preservesPartitioning=False)

• 对rdd中的每一个分区进行f操作。与map的区别是，mapPartitions是对每一个分区操作，map是对每一个元素进行操作

groupByKey()

• 原rdd为键值对，groupByKey()则将原rdd的元素相同键的值编进一个sequence

mapValues(f)

• 将键值对形式的rdd的值用函数f来作用，而保持键不变，同时分区不变。

parallelize(c, numSlice)

• 创建rdd，c可以是列表或元组（其它没试）

glom()

• 将原rdd相同分区的元素放在一个列表中构成新的rdd

rdd1 = sc.parallelize([1,2,3,4,5,6,7,8,9], 3)
rdd1.collect()
# [1, 2, 3, 4, 5, 6, 7, 8, 9]
rdd1.glom().collect()
# [[1, 2, 3], [4, 5, 6], [7, 8, 9]]

collect()

• 返回由rdd所有元素组成的列表

take(num)

• 返回前num个元素

count()

• 返回元素个数

rdd.getNumPartitions()  # 查看rdd中的分区数
rdd.count()  # 查看元素个数

报错总结

spark报错总结

• saveAsPickleFile(outPKL)，outPKL是输出文件的路径。如果路径已存在会报错Output directory already exists（防止误删文件）。

• pyspark报错java.io.IOException: Cannot run program “python”: CreateProcess error=2

  spark运行时无法找到python解释器

  在pycharm右上角点击Edit Configurations，

  在environment variable中插入环境变量

  PYTHONUNBUFFERED = 1

  PYSPARK_PYTHON = C:\Python27\python2.exe

  SPARK_HOME = C:\spark

  （远程）PYSPARK_PYTHON = /usr/bin/python

• saveAsPickleFile(outPKL)，outPKL是输出文件的路径。如果路径已存在会报错Output directory already exists（防止误删文件）。

• 中间pickle文件和原始数据pickle文件，用pickle和cpickle.load()都加载不出来，要用f = sc.pickleFile(path)，再print(f.take(num))

• java.net.SocketException: Connection reset by peer

  修改spark/python/pyspark目录下worker.py文件

  def process():为以下内容：

  def process():
  	iterator = deserializer.load_stream(infile)
    serializer.dump_stream(func(split_index, iterator), outfile)
    for obj in iterator:
       pass
  

  然后将pyspark目录打包为zip 替换掉spark/python/lib/目录下对应的pyspark.zip文件

• ERROR:py4j.java_gateway:An error occurred while trying to connect to the Java server (127.0.0.1:9992)

  更改Internet service设置：控制面板-程序-启用或关闭windows功能-Internet Information Service下全选