【博学谷IT技术支持】
Spark是一种快速、通用、可扩展的大数据分析引擎，2009年诞生，2010年开源，2013年成为Apache孵化项目，2014年成为Apache顶级项目。目前，Spark生态系统已经发展成为一个包含多个子项目的集合，其中包含SparkSQL、Spark Streaming、GraphX、MLlib等子项目，Spark是基于内存计算的大数据并行计算框架。Spark基于内存计算，提高了在大数据环境下数据处理的实时性，同时保证了高容错性和高可伸缩性，允许用户将Spark部署在大量廉价硬件之上，形成集群。PySpark 是 Spark 为 Python 开发者提供的 API以下是它的一些基本操作
- 依赖导入
from pyspark import SparkContext, SparkConf
- 初始化
sc = SparkContext(conf=conf)
conf = SparkConf().setAppName("test").setMaster("local[4]")
- 通过读取文件生成RDD
path = "hdfs://min-node1:8020/pyspark/wd/input/words.txt"
rdd1 = sc.textFile(path, 3)
- 第二个参数指定分区数
rdd = sc.parallelize(range(1, 11),2)
rdd.collect() // [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
- 多少个分区
rdd.getNumPartitions() // 2
- take操作将前若干个数据汇集到Driver，相对collect安全
part_data = rdd.take(4)
part_data // [1, 2, 3, 4]
- takesample可以随机若干个到Driver,第一个参数设置是否放回抽样
sample_data = rdd.takeSample(False, 3, 0)
sample_data // [8, 9, 2]
- first 取第一个数据
rdd.first() // 1
- count查看RDD元素数量
rdd.count() // 10
- reduce利用二元函数对数据进行规约
rdd.reduce(lambda x, y: x + y) // 55
- foreach对每一个元素执行某种操作，不生成新的RDD
accum = sc.accumulator(0)
rdd.foreach(lambda x: accum.add(x))
accum.value // 55
- countByKey按照key统计数量
pairRdd = sc.parallelize([(1, 1), (1, 4), (3, 9), (2, 16)])
pairRdd.countByKey() // defaultdict(int, {1: 2, 3: 1, 2: 1})
- saveAsTextFile保存rdd成text文件到本地
test_file = "hdfs://min-node1:8020/pyspark/txt/rdd.txt"
# rdd1 = sc.parallelize(range(5))
# rdd1.saveAsTextFile(test_file)
- 重新读入会被解析文本
rdd_loaded = sc.textFile(test_file)
rdd_loaded.collect() // ['0', '1', '2', '3', '4']
- map 操作对每一个元素进行一个映射转换
- flatMap操作执行将每个元素生成一个Array后压平
rdd3 = sc.parallelize(['hello world', 'hello china'])
print(rdd3.map(lambda x: x.split(" ")).collect()) // [['hello', 'world'], ['hello', 'china']]
print(rdd3.flatMap(lambda x: x.split(" ")).collect()) // [['hello', 'world'], ['hello', 'china']]
- filter 应用过滤条件过滤掉一些数据
rdd2 = sc.parallelize(range(10), 3)
rdd2.collect() // [0, 1, 2, 3, 4, 5, 6, 7, 8, 9]
rdd2.filter(lambda x: x > 5).collect() // [6, 7, 8, 9]
- sample 对原RDD在每个分区按照比例进行抽样，第一个参数设置是否可以重复抽样
rdd2.sample(False, 0.5, 0).collect() // [1, 3, 5]
- distinct 去重
sc.parallelize([1, 1, 2, 2, 3, 3, 4, 5]).distinct().collect() // [4, 1, 5, 2, 3]
- subtract 找到属于前一个RDD而不属于后一个RDD的元素
a = sc.parallelize(range(10))
b = sc.parallelize(range(5, 15))
c = sc.parallelize(range(10))
print(a.subtract(b).collect()) # [0, 1, 2, 3, 4]
print(a.subtract(c).collect()) # []
- union合并数据
a = sc.parallelize(range(5))
b = sc.parallelize(range(3, 8))
a.union(b).collect() # [0, 1, 2, 3, 4, 3, 4, 5, 6, 7]
- intersection 求交集
a = sc.parallelize(range(1, 6))
b = sc.parallelize(range(3, 9))
a.intersection(b).collect() // [3, 4, 5]
- cartesian 笛卡尔积
boys = sc.parallelize(["张三", "李四"])
girls = sc.parallelize(["王五", "赵六"])
boys.cartesian(girls).collect() // [('张三', '王五'), ('张三', '赵六'), ('李四', '王五'), ('李四', '赵六')]
- sortBy按照某个方式进行排序
# 指定按照第3个元素大小进行排序
rdd4 = sc.parallelize([(1, 2, 3), (3, 2, 2), (4, 1, 1)])
rdd4.sortBy(lambda x: x[2], ascending=False).collect() // [(1, 2, 3), (3, 2, 2), (4, 1, 1)]
- zip 按照拉链方式连接
# 需要两个RDD具有相同的分区， 每个分区元素数量相同
rdd_name = sc.parallelize(["三藏", "猴子", "猪八戒"])
rdd_agg = sc.parallelize([19, 18, 29])
rdd_name.zip(rdd_agg).collect() // [('三藏', 19), ('猴子', 18), ('猪八戒', 29)]
- zipWithindex 将RDD和一个从0开始的递增序列按照拉链方式连接
rdd_name = sc.parallelize(["张三", "lisi", "wangwu", "赵六"])
rdd_index = rdd_name.zipWithIndex()
rdd_index.collect() // [('张三', 0), ('lisi', 1), ('wangwu', 2), ('赵六', 3)]
- reduceByKey对相同的key对应的value应用二元归并操作
rdd = sc.parallelize([("张三", 1), ("wangwu", 2), ("张三", 4), ("wangwu", 5)])
rdd_reduceBK = rdd.reduceByKey(lambda x, y: x + y).collect()
rdd_reduceBK // [('张三', 5), ('wangwu', 7)]
- groupBykey将相同的key对应的values收集成一个Iterator
rdd_groupBK = rdd.groupByKey().collect()
rdd_groupBK
// [('张三', <pyspark.resultiterable.ResultIterable at 0x181e0fec2e0>),
// ('wangwu', <pyspark.resultiterable.ResultIterable at 0x181e1e13fa0>)]
- sortByKey 按照Key排序
rdd_sortBK = rdd.\
map((lambda res_tuple: (res_tuple[1], res_tuple[0])))\
.sortByKey()\
.map((lambda res_tuple: (res_tuple[1], res_tuple[0])))\
.collect()
rdd_sortBK // [('张三', 1), ('wangwu', 2), ('张三', 4), ('wangwu', 5)]