赞
踩
Apache Spark是专为大规模数据处理而设计的快速通用的计算引擎 。现在形成一个高速发展应用广泛的生态系统。
特点介绍
Spark 主要有三个特点:
首先,高级 API 剥离了对集群本身的关注,Spark 应用开发者可以专注于应用所要做的计算本身。
其次,Spark 很快,支持交互式计算和复杂算法。
最后,Spark 是一个通用引擎,可用它来完成各种各样的运算,包括 SQL 查询、文本处理、机器学习等,而在 Spark 出现之前,我们一般需要学习各种各样的引擎来分别处理这些需求。(来源百度百科)
park对python语言的支持--->PySpark
Spark对Python语言的支持,重点体现在Python的第三方库: PySpark
PySpark是由Spark官方开发的Python语言第三方库Python开发者可以使用pip程序快速的安装PySpark并像其它三方库那样直接使用。
在命令行中输入:pip install pyspark
使用国内代理镜像网站:pip install -i https://pypi.tuna.tsinghua.edu.cn/simple pyspark
想要使用PySpark库完成数据处理,首先需要构建一个执行环境入口对象。
PySpark的执行环境入口对象是:类SparkContext的类对象
如何通过代码获得类对象,代码如下:
- from pyspark import SparkConf, SparkContext
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
- # 打印pyspark的运行版本
- print(sc.version)
- # 停止SparkContext对象的运行(停止pyspark程序)
- sc.stop()
PySpark的编程,主要分为如下三大步骤:
PySpark支持多种数据的输入,在输入完成后,都会得到一个:RDD类的对象
RDD全称为: 弹性分布式数据集(Resilient Distributed Datasets)
PySpark针对数据的处理,都是以RDD对象作为载体即:
(1)数据存储在RDD内
(2)各类数据的计算方法,也都是RDD的成员方法
(3)RDD的数据计算方法,返回值依旧是RDD对象
PySpark支持通过SparkContext对象的parallelize成员方法,将list、tuple、set、dict、str转换为PySpark的RDD对象。
- from pyspark import SparkConf, SparkContext
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
- rdd1 = sc.parallelize([1, 2, 3, 4, 5]) # 列表
- rdd2 = sc.parallelize((1, 2, 3, 4, 5)) # 元组
- rdd3 = sc.parallelize("study python") # 字符串
- rdd4 = sc.parallelize({1, 2, 3, 4, 5}) # 集合
- rdd5 = sc.parallelize({"key1": "value1", "key2": "value2", "key3": "value3"}) # 字典
-
- # 如果要查看RDD里面有什么内容,需要用collect()方法
- print(rdd1.collect())
- print(rdd2.collect())
- print(rdd3.collect())
- print(rdd4.collect())
- print(rdd5.collect())
-
- # # 停止SparkContext对象的运行(停止pyspark程序)
- sc.stop()

注意:
(1)字符串会被拆分出1个个的字符,存入RDD对象字典
(2)仅有key会被存入RDD对象
PySpark支持通过SparkContext入口对象,来读取文件并构建出RDD对象
前面逻辑都是一样的,只是调用方法不一样,需要使用sc.textFile
- from pyspark import SparkConf, SparkContext
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
-
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
-
- rdd = sc.textFile("G:\资料\2011年1月销售数据.txt")
-
- # 如果要查看RDD里面有什么内容,需要用collect()方法
- print(rdd.collect())
-
- # # 停止SparkContext对象的运行(停止pyspark程序)
- sc.stop()
PySpark的数据计算,都是依赖RDD对象内置丰富的“成员方法(算子)”来进行的
因为spark是一个分布式程序,内部运行机制比较复杂,暂不讨论,我们只需要知道在python中运行spark程序时,需要额外增加以下代码,否则会报错:spark找不到python程序
- # 添加'PYSPARK_PYTHON'和'PYSPARK_DRIVER_PYTHON'的执行环境
- os.environ['PYSPARK_PYTHON'] = sys.executable # "E:/python/python.exe"
- os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable # "E:/python/python.exe"
功能:将RDD的数据一条条处理,返回新的RDD
语法:rdd.map(fun)
# 需要传入一个函数
需求:给[1,2,3,4,5]
每个数字乘以10
代码如下:
- import sys
- from pyspark import SparkConf, SparkContext
- import os
-
- # 添加'PYSPARK_PYTHON'和'PYSPARK_DRIVER_PYTHON'的执行环境
- os.environ['PYSPARK_PYTHON'] = sys.executable # "E:/python/python.exe"
- os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable # "E:/python/python.exe"
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
- # 需求:通过map方法将全部数据都乘以10
- rdd = sc.parallelize([1, 2, 3, 4, 5])
- # 把rdd中的每一个数据都调用匿名函数(也可通过def重新定义函数)去处理
- rdd2 = rdd.map(lambda x: x * 10)
- print(rdd2.collect())
- # 停止SparkContext对象的运行(停止pyspark程序)
- sc.stop()

功能:对RDD执行map操作,然后进行解除嵌套操作
- # 嵌套的list
- list = [[1,2,3],[4,5,6],[7,8,9]]
-
- # 解除了嵌套
- list = [1,2,3,4,5,6,7,8,9]
示例代码:
- import sys
- from pyspark import SparkConf, SparkContext
- import os
-
- # 添加'PYSPARK_PYTHON'和'PYSPARK_DRIVER_PYTHON'的执行环境
- os.environ['PYSPARK_PYTHON'] = sys.executable # "E:/python/python.exe"
- os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable # "E:/python/python.exe"
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
-
- rdd = sc.parallelize(["hello python", "hello word", "hello friend"])
- # 需求:将RDD中的每一个单词都提取出来
- rdd2 = rdd.flatMap(lambda x: x.split(" "))
- print(rdd2.collect())

功能:针对KV型RDD(二元元组),自动按照key分组,然后根据你提供的聚合逻辑,完成组内数据(value)的聚合操作。
语法:
- rdd.reduceByKey(func)
- # func:(V,V)-> V
- # 接受2个传入参数(类型要一致),返回一个返回值,类型和传入要求一致。
示例代码:
- # 前面创建RDD对象不做赘述了
- # 准备一个RDD
- rdd = sc.parallelize([('男', 99), ('男', 80), ('男', 70), ('女', 100), ('女', 85)])
- # 求男生和女生两个组的成绩之和
- rdd2 = rdd.reduceByKey(lambda a, b: a + b)
- print(rdd2.collect())
案例1
对下面txt文件中的单词进行计数统计
代码如下:
- import sys
- from pyspark import SparkConf, SparkContext
- import os
-
- # 添加'PYSPARK_PYTHON'和'PYSPARK_DRIVER_PYTHON'的执行环境
- os.environ['PYSPARK_PYTHON'] = sys.executable # "E:/python/python.exe"
- os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable # "E:/python/python.exe"
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
-
- # 读取数据文件
- rdd = sc.textFile("G:/hello.txt")
- # 读取全部单词
- word_rdd = rdd.flatMap(lambda x: x.split(" "))
- # 将所有单词都转换成二元元组,单词为key,value设置为1
- word_one_rdd = word_rdd.map(lambda word: (word, 1))
- # 分组并求和传承传承
- result_rdd = word_one_rdd.reduceByKey(lambda a, b: a + b)
- print(result_rdd.collect())

功能:过滤想要的数据进行保留
语法:
- rdd.filter(func)
- # func: (T)-->bool 传入1个参数进来随意类型,返回值必须是true or false
示例代码:
- # 前面创建RDD对象不做赘述了
- # 准备一个RDD
- rdd = sc.parallelize([1, 2, 3, 4, 5])
- # 对RDD的数据进行过滤,偶数返回true,保留偶数
- rdd2 = rdd.filter(lambda num: num % 2 == 0)
- print(rdd2.collect())
功能:对RDD数据进行去重,返回新RDD
语法:rdd.distinct()
实例代码:
- # 前面创建RDD对象不做赘述了
- # 准备一个RDD
- rdd = sc.parallelize([1, 1, 2, 2, 3, 3, 4, 5, 6, 7, 7, 8, 9])
- # 对RDD的数据进行去重
- rdd2 = rdd.distinct()
- print(rdd2.collect())
功能:对RDD数据进行排序,基于你指定的排序依据
语法:
- rdd.sortBy(func,ascending=False, numPartitions=1)
- # func: (T) -> U: 告知按照rdd中的哪个数据进行排序,比如 lambda x: [1] 表按照rdd中的第二列元素进行排序
- # ascending True升序 False 降序
- # numPartitions: 用多少分区排序
示例代码:
以上对单词进行计数统计的案例1中,对结果按照单词出现的次数从大到小进行排序
- # 对结果进行排序
- final_rdd = result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
- print(final_rdd.collect())
案例2
需求:对以下文件使用spark读取文件进行计算:
(1)各个城市销售额排名,从大到小
(2)全部城市,有哪些商品类别在售卖
(3)北京市有哪些商品类别在售卖
案例代码:
- import sys
- from pyspark import SparkConf, SparkContext
- import os
- import json
-
- # 添加'PYSPARK_PYTHON'和'PYSPARK_DRIVER_PYTHON'的执行环境
- os.environ['PYSPARK_PYTHON'] = sys.executable # "E:/python/python.exe"
- os.environ['PYSPARK_DRIVER_PYTHON'] = sys.executable # "E:/python/python.exe"
-
- # 创建SparkConf类对象
- conf = SparkConf().setMaster("local[*]").setAppName("test_spark_app") # 链式调用,不管调用什么样的方法,返回的都是同一个对象
- # 基于SparkConf类对象创建SparkContext对象
- sc = SparkContext(conf=conf)
-
- # (1)城市销售额排名
- # 读取数据文件
- file_rdd = sc.textFile("G:/orders.txt")
- # 读取文件中单个json字符串
- json_str_rdd = file_rdd.flatMap(lambda x: x.split("|"))
- # 将单个字符串转换为字典
- dict_rdd = json_str_rdd.map(lambda x: json.loads(x))
- # 取出城市和销售额数据(城市,销售额)
- city_money_rdd = dict_rdd.map(lambda x: (x['areaName'], int(x['money'])))
- # 按照城市分组,按照销售额累计
- city_result_rdd = city_money_rdd.reduceByKey(lambda a, b: a + b)
- # 按照销售额累计结果进行排序
- result1_rdd = city_result_rdd.sortBy(lambda x: x[1], ascending=False, numPartitions=1)
- print("各个城市销售额排名,从大到小:", result1_rdd.collect())
- # (2)全部城市,有哪些商品类别在售卖
- category_rdd = dict_rdd.map(lambda x: x['category']).distinct()
- print("全部城市,有哪些商品类别在售卖:", category_rdd.collect())
- # (3)北京市有哪些商品类别在售卖
- beijing_data_rdd = dict_rdd.filter(lambda x: x['areaName'] == '北京')
- beijing_category = beijing_data_rdd.map(lambda x: x['category']).distinct()
- print('北京市有哪些商品类别在售卖:', beijing_category.collect())

RDD的结果输出为Python对象的各类方法
功能:将RDD各个分区内的数据,统一收集到Driver中,形成一个List对象
语法:rdd.collect()
,返回值是一个list
此方法我们前面一直在使用,不做赘述
功能:对RDD数据集按照你传入的逻辑进行聚合
语法:rdd.reduce(func)
,两参数传入,1个返回值,返回值和参数要求类型一致
示例代码:
- rdd = sc.parallelize([1, 2, 3, 4, 5])
- print(rdd.reduce(lambda a, b: a + b))
- # 输出结果为两两相加的值:15
功能:取RDD的前n个元素,组成list返回
语法:rdd.take(num)
,num代表前几个元素
示例代码:
- rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
- print(rdd.take(5))
- # 输出结果为:[1, 2, 3, 4, 5]
功能:计算RDD有多少条数据,返回值是一个数字
用法:rdd.count()
示例代码:
- rdd = sc.parallelize([1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
- print(rdd.count())
- # 输出结果为:10
注意:
关于Spark的方法(算子)还有很多很多,我们目前不是深入学习python语言,同时也没有学习分布式,只是简单学习对自动化测试打下基础,所以本篇文章只介绍了最基础的几个。
将RDD的内容输出到文件中
功能:将RDD的数据写入文本文件中,支持本地写出,hdfs(Hadoop分布式文件系统)等文件系统。
语法: rdd.saveAsTextFile("../data/output/test.txt")
想要这个方法正常运行,还需要配置Hadoop依赖,自行百度配置
运行之后,内容会存在多个分区中,输出的结果是一个文件夹,有几个分区就输出多少个结果文件
修改RDD分区为1个
方式1:SparkConf对象设置conf.set("spark.default.parallelism", "1")
方式2:创建RDD的时候,sc.parallelize方法传入numSlices参数为1
- # 方式1,SparkConf对象设置属性全局并行度为1:
- conf = SparkConf().setMaster("Tocal[*]").setAppName("test_spark")
- conf.set("spark.default.parallelism", "1")
- sc = SparkContext(conf=conf)
- # 方式2, 创建RDD的时候设置(parallelize方法传入numSlices参数为1)
- rdd1 = sc.parallelize([1, 2, 3, 4, 5], numSlices=1)
- rdd1 = sc.parallelize([1, 2, 3, 4, 5], 1)
行动吧,在路上总比一直观望的要好,未来的你肯定会感 谢现在拼搏的自己!如果想学习提升找不到资料,没人答疑解惑时,请及时加入扣群: 320231853,里面有各种软件测试+开发资料和技术可以一起交流学习哦。
最后感谢每一个认真阅读我文章的人,礼尚往来总是要有的,虽然不是什么很值钱的东西,如果你用得到的话可以直接拿走:
这些资料,对于【软件测试】的朋友来说应该是最全面最完整的备战仓库,这个仓库也陪伴上万个测试工程师们走过最艰难的路程,希望也能帮助到你!
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。