大数据分析与应用实验任务九

news/2024/11/18 4:28:54/

大数据分析与应用实验任务九

实验目的

  • 进一步熟悉pyspark程序运行方式；

  • 熟练掌握pysaprkRDD基本操作相关的方法、函数，解决基本问题。

实验任务

进入pyspark实验环境，打开命令行窗口，输入pyspark，完成下列任务：

在实验环境中自行选择路径新建以自己姓名拼音命名的文件夹，后续代码中涉及的文件请保存到该文件夹下（需要时文件夹中可以创建新的文件夹）。

一、参考书中相应代码，练习RDD持久性、分区及写入文件（p64、67、80页相应代码）。
1.持久化

迭代计算经常需要多次重复使用同一组数据。下面就是多次计算同一个RDD的例子。

listlzy=["Hadoop","Spark","Hive","Darcy"]
rddlzy=sc.parallelize(listlzy)
print(rddlzy.count())#行动操作，触发一次真正从头到尾的计算
print(','.join(rddlzy.collect()))#行动操作，触发一次真正从头到尾的计算

image-20231123112954850

一般而言，使用cache()方法时，会调用persist(MEMORY_ONLY)。针对上面的实例，增加持久化语句以后的执行过程如下：

listlzy=["Hadoop","Spark","Hive","Darcy"]
rdd=sc.parallelize(listlzy)
rdd.cache()#会调用persist(MEMORY_ONLY)，但是，语句执行到这里，并不会缓存rdd，因为这时rdd还没有被计算生成
print(rdd.count())#第一次行动操作，触发一次真正从头到尾的计算，这时上面的rdd.cache()才会被执行，把这个rdd放到缓存中
print(','.join(rdd.collect()))#第二次行动操作，不需要触发从头到尾的计算，只需要重复使用上面缓存中的rdd

image-20231123113247048

2.分区

  • 设置分区的个数

    在调用textFile()和parallelize()方法的时候手动指定分区个数即可，语法格式如下：

    sc.textFile(path,partitionNum)

    其中，path参数用于指定要加载的文件的地址，partitionNum参数用于指定分区个数。下面是一个分区的实例。

    listlzy=[5,2,0,1,3,1,4]
rddlzy=sc.parallelize(listlzy,2)//设置两个分区

    image-20231123113420685

  • 使用repartition方法重新设置分区个数

    通过转换操作得到新RDD时，直接调用repartition方法即可。例如

datalzy=sc.parallelize([1,2,3,4,5],2)
len(datalzy.glom().collect())#显示datalzy这个RDD的分区数量
rdd = datalzy.repartition(1) #对 data 这个 RDD 进行重新分区
len(rdd.glom().collect()) #显示 rdd 这个 RDD 的分区数量

image-20231123113519540

  • 自定义分区方法

    下面是一个实例，要求根据 key 值的最后一位数字将 key 写入到不同的文件中，比如，10 写入到 part-00000，11 写入到 part-00001，12 写入到 part-00002。打开一个 Linux 终端，使用 vim 编辑器创建一个代码文件“/root/Desktop/luozhongye/TestPartitioner.py”，输入以下代码：

    from pyspark import SparkConf, SparkContextdef MyPartitioner(key):print("MyPartitioner is running")print('The key is %d' % key)return key % 10def main():print("The main function is running")conf = SparkConf().setMaster("local").setAppName("MyApp")sc = SparkContext(conf=conf)data = sc.parallelize(range(10), 5)data.map(lambda x: (x, 1)).partitionBy(10, MyPartitioner).map(lambda x: x[0]).saveAsTextFile("file:///root/Desktop/luozhongye/partitioner")if __name__ == '__main__':main()

使用如下命令运行 TestPartitioner.py：

cd /root/Desktop/luozhongye
python3 TestPartitioner.py

或者，使用如下命令运行 TestPartitioner.py：

cd /root/Desktop/luozhongye
spark-submit TestPartitioner.py

程序运行后会返回如下信息：

image-20231123114351343

3.文件数据写入
  • 把 RDD 写入到文本文件中
textFile = sc.textFile("file:///root/Desktop/luozhongye/wordlzy.txt") 
textFile.saveAsTextFile("file:///root/Desktop/luozhongye/writeback")

其中wordlzy.txt的内容

Hadoop is good
Spark is fast
Spark is better
luozhongye is handsome

image-20231123115053912

Spark 采用惰性机制。可以使用如下的“行动”类型的操作查看 textFile 中的内容：

textFile.first()

正因为 Spark 采用了惰性机制，在执行转换操作的时候，即使输入了错误的语句，pyspark 也不会马上报错，而是等到执行“行动”类型的语句启动真正的计算时，“转换”操作语句中的错误才会显示出来，比如：

textFile = sc.textFile("file:///root/Desktop/luozhongye/wordcount/word123.txt")

image-20231123115210529

  • 把 RDD 写入到文本文件中

可以使用 saveAsTextFile()方法把 RDD 中的数据保存到文本文件中。下面把 textFile 变量中的内容再次写回到另外一个目录 writeback 中，命令如下：

textFile = sc.textFile("file:///root/Desktop/luozhongye/wordlzy.txt") 
textFile.saveAsTextFile("file:///root/Desktop/luozhongye/writeback")

进入到“/root/Desktop/luozhongye/writeback”目录查看

cd /root/Desktop/luozhongye/writeback
ls

image-20231123115411983

二、逐行理解并运行4.4.2实例“文件排序”。

新建多个txt文件file1.txt 、file2.txt 、file3.txt ，其内容分别如下：

33
37
12
40

4
16
39
5

1
45
25

要求读取所有文件中的整数，进行排序后，输出到一个新的文件中，输出的内容为每行两个整数，第一个整数为第二个整数的排序位次，第二个整数为原待排序的整数。

实现上述功能的代码文件“/root/Desktop/luozhongye/FileSort.py”的内容如下：

#!/usr/bin/env python3 
from pyspark import SparkConf, SparkContextindex = 0def getindex():global indexindex += 1return indexdef main():conf = SparkConf().setMaster("local[1]").setAppName("FileSort")sc = SparkContext(conf=conf)lines = sc.textFile("file:///root/Desktop/luozhongye/file*.txt")index = 0result1 = lines.filter(lambda line: (len(line.strip()) > 0))result2 = result1.map(lambda x: (int(x.strip()), ""))result3 = result2.repartition(1)result4 = result3.sortByKey(True)result5 = result4.map(lambda x: x[0])result6 = result5.map(lambda x: (getindex(), x))result6.foreach(print)result6.saveAsTextFile("file:///root/Desktop/luozhongye/sortresult")if __name__ == '__main__':main()

image-20231123232005765

三、完成p96实验内容3，即“编写独立应用程序实现求平均值问题”，注意每位同学自己修改题目中的数据。

每个输入文件表示班级学生某个学科的成绩，每行内容由两个字段组成，第一个字段是学生名字，第二个字段是学生的成绩；编写 Spark 独立应用程序求出所有学生的平均成绩，并输出到一个新文件中。

数学成绩.txt：

小罗 110
小红 107
小新 100
小丽 99

英语成绩.txt：

小罗 95 
小红 81 
小新 82
小丽 76

政治成绩.txt:

小罗 65 
小红 71 
小新 61 
小丽 66

408成绩.txt：

小罗 100
小红 103
小新 94
小丽 110

实现代码如下：

from pyspark import SparkConf, SparkContext# 初始化Spark配置和上下文
conf = SparkConf().setAppName("AverageScore")
sc = SparkContext(conf=conf)# 读取数学成绩文件
math_rdd = sc.textFile("数学成绩.txt").map(lambda x: (x.split()[0], int(x.split()[1])))# 读取英语成绩文件
english_rdd = sc.textFile("英语成绩.txt").map(lambda x: (x.split()[0], int(x.split()[1])))# 读取政治成绩文件
politics_rdd = sc.textFile("政治成绩.txt").map(lambda x: (x.split()[0], int(x.split()[1])))# 读取408成绩文件
computer_rdd = sc.textFile("408成绩.txt").map(lambda x: (x.split()[0], int(x.split()[1])))# 合并所有成绩数据
all_scores_rdd = math_rdd.union(english_rdd).union(politics_rdd).union(computer_rdd)# 计算每个学生的成绩总和和成绩数量
sum_count_rdd = all_scores_rdd.combineByKey(lambda value: (value, 1),lambda acc, value: (acc[0] + value, acc[1] + 1),lambda acc1, acc2: (acc1[0] + acc2[0], acc1[1] + acc2[1]))# 计算平均成绩
average_scores_rdd = sum_count_rdd.mapValues(lambda x: x[0] / x[1])# 输出到新文件
average_scores_rdd.saveAsTextFile("平均成绩")# 关闭Spark上下文
sc.stop()

image-20231123233508387

实验心得

在这次实验中，我进一步熟悉了使用PySpark进行大数据处理和分析的方法，并深入了解了PySpark RDD的基本操作。学会了分区、持久化、数据写入文件，并解决实际问题。这次实验让我对PySpark有了更深入的理解，并增强了我处理和分析大数据的能力。


http://www.ppmy.cn/news/1236954.html

相关文章

通过这个简单的技巧让我们的 JavaScript 代码变得异常快

通过这个简单的技巧让我们的 JavaScript 代码变得异常快

通过这个简单的技巧让我们的 JavaScript 代码变得异常快 秘诀：了解JavaScript 虚拟机(VM)的内部工作原理。 首先，我们来谈谈像 V8 这样的JavaScript 虚拟机(VM)。可以把它想象成我们的操作的大脑 —— 它将我们简洁的代码变成计算机可以理解和执行的东…
阅读更多...
QT 中的元对象系统

QT 中的元对象系统

作为一名十几年的 C 程序员，最近一段时间使用 QT 开发程序，发现 QT 中还是有许多值得深入理解的技术。QT 不仅仅是一个应用程序开发框架，还有一些对标准 C 的扩充。本文和大家一起探讨 QT 中的元对象系统。 在分析 QT 中的元对象系统之前&…
阅读更多...
百战python02-语言元素

百战python02-语言元素

文章目录 指令与程序变量与类型变量命名变量的使用运算符赋值运算符比较运算符和逻辑运算符练习1:华氏温度转换为摄氏温度练习2：输入圆的半径计算计算周长和面积练习3：输入年份判断是不是闰年字符串常用操作注：需要对python有基本了解，可查看本作者python基础专栏，有任何问…
阅读更多...
洛谷 P1883 函数

洛谷 P1883 函数

P1883 函数 - 洛谷 | 计算机科学教育新生态 (luogu.com.cn) Error Curves - 洛谷 | 计算机科学教育新生态 (luogu.com.cn) 这两题是一模一样的，过一题水两题。 分析 主要难点在于证明F(x)是一个单峰函数可以被三分，但是我随便画了几个f(x)之后发现好像…
阅读更多...
webpack打包三方库直接在html里面使用

webpack打包三方库直接在html里面使用

场景：我是小程序中使用wxmp-rsa库进行加密，然后在html里面解密 我就想把wxmp-rsa库打包到一个js里面，然后在html里面直接引入使用。 webpack配置 const path require("path"); const MiniCssExtractPlugin require("mini-…
阅读更多...
Linux进程通信——信号（一）

Linux进程通信——信号（一）

原理 对于 Linux来说，实际信号是软中断，许多重要的程序都需要处理信号。 信号，为 Linux 提供了一种处理异步事件的方法。比如，终端用户输入了ctrlc来中断程序，会通过信号机制停止一个程序。 概述 信号的名字和编号 …
阅读更多...
目标检测YOLO系列从入门到精通技术详解100篇-【目标检测】SLAM（基础篇）（三）

目标检测YOLO系列从入门到精通技术详解100篇-【目标检测】SLAM（基础篇）（三）

目录 前言 移动机器人视觉SLAM回环检测 01 回环检测问题描述 02 主流回环检测方法 2.1 根据路标点先验信息
阅读更多...
【LeetCode刷题-链表】--61.旋转链表

【LeetCode刷题-链表】--61.旋转链表

61.旋转链表 方法： 记给定的链表的长度为n,注意当向右移动的次数k>n时，仅需要向右移动k mod n次即可，因为每n次移动都会让链表变为原状 将给定的链表连接成环，然后将指定位置断开 /*** Definition for singly-linked list.*…
阅读更多...
最新文章

Copyright @ 2022~2023