pyspark常用的基本操作

墨蓝 2023-06-22 04:28 0阅读 0赞

#### 一.通过外部数据创建RDD ####

# 1.通过本地数据创建RDD
    # 准备本地文件
    cat myspark.txt 
    >hello my name is Cocktail_py
    >welcome to my blogs
    
    # pyspark shell
    rdd = sc.textFile("file:root/myspark.txt")
    rdd.collect()
    >['hello my name is Cocktail_py', 'welcome to my blogs']
    
    # 2.通过HDFS数据创建RDD
    hadoop fs -put myspark.txt /
    hadoop fs -cat /myspark.txt
    >hello my name is Cocktail_py
    >welcome to my blogs
    
    # pyspark shell
    rdd = sc.textFile("hdfs://192.168.18.126:9000/myspark.txt")
    rdd.collect()
    ['hello my name is Cocktail_py', 'welcome to my blogs']

#### 二.Transformation算子 ####

# transformation操作都是惰性的,不会立即计算结果
    
    # 1.map(f, preservesPartitioning=False) 将函数作用到数据集的每一个元素上,生成一个新的RDD
    # 利用parallelize 将传入的已有的可迭代对象或集合创建RDD,可指定分区数,即利用多少线程进行计算
    rdd = sc.parallelize([7,7,7,1,2,3],3)
    rdd.map(lambda x:x+1 )
    >PythonRDD[80] at RDD at PythonRDD.scala:53
    
    rdd1 = rdd.map(lambda x:x+1 )
    rdd1.collect()
    >[8, 8, 8, 2, 3, 4]
    
    # 2.filter(f) 过滤函数中为false对象,生成新的rdd
    rdd = sc.parallelize([True,True,False,True,False,True,True])
    rdd.filter(lambda x:x)
    >PythonRDD[83] at RDD at PythonRDD.scala:53
    
    rdd1 = rdd.filter(lambda x:x)
    rdd1.collect()
    >[True, True, True, True, True]
    
    # 3.flatMap(f, preservesPartitioning=False) 首先将一个函数应用到这个RDD的所有元素上，然后将结果压扁,返回一个新的RDD
    rdd = sc.parallelize([2, 3, 4])
    sorted(rdd.flatMap(lambda x: range(1, x)).collect())
    >[1, 1, 1, 2, 2, 3]
    sorted(rdd.flatMap(lambda x: [(x, x), (x, x)]).collect())
    >[(2, 2), (2, 2), (3, 3), (3, 3), (4, 4), (4, 4)]
    
    rdd = sc.parallelize(['hello my name is Cocktail_py', 'welcome to my blogs'])
    rdd.flatMap(lambda x:x.split(" ")).collect()
    >['hello', 'my', 'name', 'is', 'Cocktail_py', 'welcome', 'to', 'my', 'blogs']
    # Map与flatMap的区别
    rdd.map(lambda x:x.split(" ")).collect()
    >[['hello', 'my', 'name', 'is', 'Cocktail_py'], ['welcome', 'to', 'my', 'blogs']]
    
    # 4.union(other) 返回此RDD和另一个RDD的并集。
    rdd = sc.parallelize([1, 1, 2, 3])
    rdd.union(rdd).collect()
    >[1, 1, 2, 3, 1, 1, 2, 3]
    
    # 5.intersection(other) 返回这个RDD和另一个RDD的交集。即使输入RDDs包含任何重复的元素，输出也不会包含任何重复的元素。
    rdd1 = sc.parallelize([1, 10, 2, 3, 4, 5])
    rdd2 = sc.parallelize([1, 6, 2, 3, 7, 8])
    rdd1.intersection(rdd2).collect()
    >[1, 2, 3]
    # 6.mapValues(f) 将一个函数作用于键值对(key-value pair)形式的RDD 的 每个值(value)上面,没有改变键(keys)
    x = sc.parallelize([("a", ["apple", "banana", "lemon"]), ("b", ["grapes"])])
    def f(x): return len(x)
    x.mapValues(f).collect()
    >[('a', 3), ('b', 1)]
    # 7.groupByKey(numPartitions=None, partitionFunc=<function portable_hash>) 将RDD中每个键的值分组为单个序列
    # 如果您是为了对每个键执行聚合(例如求和或平均值)而进行分组，那么使用reduceByKey或aggregateByKey将提供更好的性能。
    rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
    rdd1 = rdd.groupByKey()
    rdd1.collect()
    >[('b', <pyspark.resultiterable.ResultIterable object at 0x7f25a96c76d8>), ('a', <pyspark.resultiterable.ResultIterable object at 0x7f25a96c77f0>)]
    # 查看元组对象中的值
    result = sorted(rdd1.collect())
    len(result[0][1])
    >2
    len(result[1][1])
    >1
    list(result[0][1])
    >[1, 1]
    list(result[1][1])
    >[1]
    sorted(rdd1.mapValues(len).collect())
    >[('a', 2), ('b', 1)]                                                           
    sorted(rdd1.mapValues(list).collect())
    >[('a', [1, 1]), ('b', [1])]
    
    # 8.reduceByKey(func, numPartitions=None, partitionFunc=<function portable_hash>) 使用关联和可交换的归约函数合并每个键的值。
    from operator import add
    rdd = sc.parallelize([("a", 1), ("b", 1), ("a", 1)])
    sorted(rdd.reduceByKey(add).collect())
    >[('a', 2), ('b', 1)]
    # rdd.reduceByKey(lambda x,y:x+y)
    
    # 9.sortByKey(ascending=True, numPartitions=None, keyfunc=<function RDD.<lambda>>)[source] 对这个RDD排序，假定它由(键、值)对组成。
    tmp = [('a', 1), ('b', 2), ('1', 3), ('d', 4), ('2', 5)]
    sc.parallelize(tmp).sortByKey().first()
    >('1', 3)
    sc.parallelize(tmp).sortByKey(True, 1).collect()
    >[('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
    sc.parallelize(tmp).sortByKey(True, 2).collect()
    >[('1', 3), ('2', 5), ('a', 1), ('b', 2), ('d', 4)]
    tmp2 = [('Mary', 1), ('had', 2), ('a', 3), ('little', 4), ('lamb', 5)]
    tmp2.extend([('whose', 6), ('fleece', 7), ('was', 8), ('white', 9)])
    sc.parallelize(tmp2).sortByKey(True, 3, keyfunc=lambda k: k.lower()).collect()
    >[('a', 3), ('fleece', 7), ('had', 2), ('lamb', 5),...('white', 9), ('whose', 6)]

#### 3.Action算子 ####

# 1.collect() 返回包含此RDD中的所有元素的列表
    
    # 2.count() Return the number of elements in this RDD.
    sc.parallelize([2, 3, 4]).count()
    >3
    # 3.take(num) 取出RDD的前num个元素
    sc.parallelize([2, 3, 4, 5, 6]).cache().take(2)
    >[2, 3]
    sc.parallelize([2, 3, 4, 5, 6]).take(10)
    >[2, 3, 4, 5, 6]
    sc.parallelize(range(100), 100).filter(lambda x: x > 90).take(3)
    >[91, 92, 93]
    
    # 4.reduce(f) 使用指定的交换和关联二进制操作符减少此RDD的元素。
    from operator import add
    sc.parallelize([1, 2, 3, 4, 5]).reduce(add)
    # sc.parallelize([1, 2, 3, 4, 5]).reduce(lambda x,y : x+y)
    >15
    sc.parallelize((2 for _ in range(10))).map(lambda x: 1).cache().reduce(add)
    >10

参考:https://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=flatmap