Spark笔记7-RDD持久化和分区

list = ["hadoop", "spark", "hive"]
rdd = sc.parallelize(list)   # 生成RDD
rdd.cache()   # 标记为持久化
print(rdd.count())   # 第一个行动化操作。触发从头到尾的计算，将结果存入缓存中
print(','.join(rdd.collect()))  # 使用上面缓存的结果，不必再次从头到尾的进行计算，使用缓存的RDD

list = [1,2,3,4]
rdd = sc.parallelize(list,4)  # 设置4个分区

data = sc.parallelize([1,2,3,4], 4) # 指定4个分区
len(data.glom().collect())  # 显示分区数目
rdd = data.repartition(2)  # 	重新设置分区数目为2

# demo.py

from pyspark import SparkConf, SparkContext

def myPartitioner(key):
  print("mypartitioner is running")
  print("the key is %d" %key)
  return key%10.

def main():
  conf = SparkConf().setMaster("local").setAppName("myapp")
  sc = SparkContext(conf=conf)  # 生成对象，就是指挥官
  data = sc.parallelize(range(10), 5)  # 分成5个分区
  data.map(lambda x: (x,1)) \ # 生成键值对，下图1
  .partitionBy(10, myPartitioner)  \ # 函数只接受键值对作为参数，将上面的data变成键值对形式传进来
  .map(lambda x:x[0])  \ # 	取出键值对的第一个元素，下图2
  .saveAsTextFile("file:///usr/local/spark/mycode/rdd/partitioner")  \ # 写入目录地址，生成10个文件


if __name__ == "__main__":
  main()