【Spark】【大数据技术基础】课程实验七 Spark基础编程实验

🕗 发布于 2024-11-25 15:19 大数据 spark 分布式

实验七：Spark初级编程实践

一、实验目的

掌握使用 Spark 访问本地文件和 HDFS 文件的方法

掌握 Spark 应用程序的编写、编译和运行方法

二、实验平台

操作系统：Ubuntu16.04

Spark版本：2.1.0

scala版本：2.11.8

Hadoop版本：2.7.1

三、实验步骤

1. 准备工作（镜像已包含，不用安装）

（1）安装spark及其API程序

安装 spark：Ubuntu下安装Spark2.11.8

安装 sbt：Ubuntu下为Spark安装配置sbt

（2）配置环境变量

修改你的 .bashrc 文件：

vim ~/.bashcrc

添加以下内容至文件顶部：

export SPARK_HOME=/usr/local/spark

export PATH=$PATH:$SPARK_HOME/bin

使修改立即生效：

source ~/.bashrc

（3）启动Hadoop

进入 Hadoop 安装目录：

cd /usr/local/hadoop

启动并使用 jps 检查结点：

./sbin/start-dfs.sh

jps

2. Spark读取文件系统的数据

（1）在spark-shell中读取Linux系统本地文件“/home/hadoop/test.txt”，然后统计出文件的行数；

创建 test.txt：

echo -e "Hello\nThis is a test\nBye!" >> ~/test.txt

启动 spark-shell：

cd /usr/local/spark

./bin/spark-shell

Scala 命令：

val textFile=sc.textFile("file:///home/hadoop/test.txt")

textFile.count()

输出如下：

scala> val textFile=sc.textFile("file:///home/hadoop/test.txt")

textFile: org.apache.spark.rdd.RDD[String] = file:///home/hadoop/test.txt MapPartitionsRDD[1] at textFile at <console>:23

scala> textFile.count()

res0: Long = 3

（2）在spark-shell中读取HDFS系统文件“/user/hadoop/test.txt”（如果该文件不存在，请先创建），然后，统计出文件的行数；

上传 test.txt 文件至 HDFS 中（终端中执行，退出 spark-shell）：

/usr/local/hadoop/bin/hdfs dfs -put ~/test.txt

Scala 命令如下（spark-shell）：

./bin/spark-shell

val textFile=sc.textFile("hdfs://localhost:9000/user/hadoop/test.txt")

textFile.count()

输出如下：

scala> val textFile=sc.textFile("hdfs://localhost:9000/user/hadoop/test.txt")

textFile: org.apache.spark.rdd.RDD[String] = hdfs://localhost:9000/user/hadoop/test.txt MapPartitionsRDD[3] at textFile at <console>:23

scala> textFile.count()

res1: Long = 3

（3）编写独立应用程序（推荐使用Scala语言），读取HDFS系统文件“/user/hadoop/test.txt”（如果该文件不存在，请先创建），然后，统计出文件的行数；通过sbt工具将整个应用程序编译打包成 JAR包，并将生成的JAR包通过 spark-submit 提交到 Spark 中运行命令。

进入 spark 安装目录：

cd /usr/local/spark

mkdir mycode && cd mycode

创建 HDFStest 目录并编写 Scala 文件：

mkdir -p HDFStest/src/main/scala

vim ./HDFStest/src/main/scala/HDFStest.scala

代码如下：

/* HDFStest.scala */

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf



object HDFStest {

    def main(args: Array[String]) {

        val logFile = "hdfs://localhost:9000/user/hadoop/test.txt"

        val conf = new SparkConf().setAppName("Simple Application")

        val sc = new SparkContext(conf)

        val logData = sc.textFile(logFile, 2)

        val num = logData.count()

        printf("The num of this file is %d\n", num)

    }

}

进入 HDFStest 目录，创建 simple.sbt：

cd HDFStest

vim simple.sbt

内容如下：

name := "A Simple HDFS Test"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

注意这里的 scalaVersion 是你的 Scala 版本，spark-core 是你的 spark 版本。

接下来，可以通过如下代码将整个应用程序打包成 JAR：

/usr/local/sbt/sbt package

打包成功输出如下：

运行如下代码使用生成的 jar 包：

/usr/local/spark/bin/spark-submit  --class  "HDFStest" /usr/local/spark/mycode/HDFStest/target/scala-2.12/a-simple-hdfs-test_2.12-1.0.jar 2>& 1 | grep The

输出如下：

3. 编写独立应用程序实现数据去重

对于两个输入文件 A 和 B，编写 Spark 独立应用程序（推荐使用 Scala 语言），对两个文件进行合并，并剔除其中重复的内容，得到一个新文件 C。下面是输入文件和输出文件的一个样例，供参考。

输入文件 A 的样例如下：

20170101 x

20170102 y

20170103 x

20170104 y

20170105 z

20170106 z

输入文件 B 的样例如下：

20170101 y

20170102 y

20170103 x

20170104 z

20170105 y

根据输入的文件 A 和 B 合并得到的输出文件 C 的样例如下：

20170101 x

20170101 y

20170102 y

20170103 x

20170104 y

20170104 z

20170105 y

20170105 z

20170106 z

进入到 mycode 目录，新建 RemDup 目录：

cd /usr/local/spark/mycode

mkdir -p RemDup/src/main/scala

cd RemDup

新建 datas 目录，写入文件 A 和文件 B：

mkdir datas

注意这里 A 和 B 文件内容不能有多余的换行符或者空格！因为今年是2024年，所以全部前缀改成2024：

vim ./datas/A

vim ./datas/B

编写 Scala 文件：

vim ./src/main/scala/RemDup.scala

代码如下：

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

import org.apache.spark.HashPartitioner



object RemDup {

    def main(args: Array[String]) {

        val conf = new SparkConf().setAppName("RemDup")

        val sc = new SparkContext(conf)

        val dataFile = "file:///usr/local/spark/mycode/RemDup/datas"

        val data = sc.textFile(dataFile,2)

        val res = data.filter(_.trim().length>0).map(line=>(line.trim,"")).partitionBy(new HashPartitioner(1)).groupByKey().sortByKey().keys

        res.saveAsTextFile("file:///usr/local/spark/mycode/RemDup/result")

    }

}

编写 simple.sbt 文件：

vim simple.sbt

内容如下：

name := "Remove Duplication"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

【注意】：scalaVersion 和spark-core改成自己的版本：

使用如下命令打包：

/usr/local/sbt/sbt package

使用生成的 jar 包：

/usr/local/spark/bin/spark-submit --class "RemDup"  /usr/local/spark/mycode/RemDup/target/scala-2.11/remove-duplication_2.11-1.0.jar

改成自己使用的版本（2.11）：

使用如下命令查看输出：

cat result/*

输出如下：

4. 编写独立应用程序实现求平均值问题

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

Algorithm 成绩：

小明 92

小红 87

小新 82

小丽 90

Database 成绩：

小明 95

小红 81

小新 89

小丽 85

Python 成绩：

小明 82

小红 83

小新 94

小丽 91

平均成绩如下：

(小红,83.67)

(小新,88.33)

(小明,89.67)

(小丽,88.67)

进入到 mycode 目录，新建 AvgScore 目录：

cd /usr/local/spark/mycode

mkdir -p AvgScore/src/main/scala

cd AvgScore

新建 datas 目录，写入文件 algorithm、database、python：

mkdir datas

注意这里 algorithm、database 和 python 文件内容不能有多余的换行符或者空格！

vim ./datas/algorithm

vim ./datas/database

vim ./datas/python

编写 Scala 文件：

vim ./src/main/scala/AvgScore.scala

代码如下：

import org.apache.spark.SparkContext

import org.apache.spark.SparkContext._

import org.apache.spark.SparkConf

import org.apache.spark.HashPartitioner



object AvgScore {

    def main(args: Array[String]) {

        val conf = new SparkConf().setAppName("AvgScore")

        val sc = new SparkContext(conf)

        val dataFile = "file:///usr/local/spark/mycode/AvgScore/datas"

        val data = sc.textFile(dataFile,3)



       val res = data.filter(_.trim().length>0).map(line=>(line.split(" ")(0).trim(),line.split(" ")(1).trim().toInt)).partitionBy(new HashPartitioner(1)).groupByKey().map(x => {

            var n = 0

        var sum = 0.0

        for(i <- x._2){

sum = sum + i

        n = n +1

         }

        val avg = sum/n

         val format = f"$avg%1.2f".toDouble

         (x._1,format)

    })

       res.saveAsTextFile("file:///usr/local/spark/mycode/AvgScore/result")

    }

}

编写 simple.sbt 文件：

vim simple.sbt

内容如下：

name := "Average Score"

version := "1.0"

scalaVersion := "2.11.8"

libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"

【注意】：scalaVersion 和spark-core改成自己的版本：

使用如下命令打包：

/usr/local/sbt/sbt package

使用生成的 jar 包：

/usr/local/spark/bin/spark-submit --class "AvgScore"  /usr/local/spark/mycode/AvgScore/target/scala-2.11/average-score_2.11-1.0.jar

改成自用版本！

使用如下命令查看输出：

cat result/*

输出如下：

四、实验总结

在实验过程中，我遇到了以下问题并将其解决：

环境配置问题：最初在配置环境变量时，由于路径错误导致无法启动Spark Shell。通过检查并修正.bashrc文件中的SPARK_HOME路径，我解决了这个问题。
- 程序编译错误：在编写数据去重程序时，由于对Scala语法不熟导致编译错误。通过查阅文档和示例代码，我逐步修正了代码中的语法错误。
  - 运行时性能问题：在处理较大文件时，程序运行缓慢。通过增加Spark应用程序的分区数和优化代码逻辑，我提高了程序的运行效率。

通过本次实验，我不仅掌握了Spark的基本操作和编程技巧，还学会了如何调试和优化Spark应用程序。实验过程中遇到的问题和挑战也加深了我对Spark工作原理的理解。此外，实验还提高了我解决实际问题的能力，为我后续的大数据技术学习和应用打下了坚实的基础。

原文地址：https://blog.csdn.net/m0_73972962/article/details/143739051

免责声明：本站文章内容转载自网络资源，如本站内容侵犯了原著者的合法权益，可联系本站删除。更多内容请关注自学内容网（zxcms.com）！

上一篇：WPF——ICON按钮制作
下一篇：jar包解压和重新打包

Mybatis-Day3
定义与SQL映射文件同名的Mapper接口，并且将Mapper接口和SQL映射文件放置在同一目录下设置SQL映射我呢见的namespace属性为Mapper接口的全限定名在Mapper接口中定义方法，
阅读更多2024-11-26
TCL大数据面试题及参考答案
HBase 中的 Region 是数据存储和管理的基本单元，Region 切分是指将一个大的 Region 按照一定的规则拆分成多个较小的 Region 的过程。Clickhouse 的物化视图是一种
阅读更多2024-11-26
选择使用whisper.cpp进行语音转文字
当你添加本地文件后，不管文件多大，数量有几个，它会马上启动文件上传，让你误以为它能够无限制的免费帮你进行转换，最差情况，估计也就是让你这种白嫖党多等一会儿。需要将一些wav格式的语音文件转成文字（AS
阅读更多2024-11-26
【MATLAB源码-第224期】基于matlab的快跳频系统仿真采用4FSK，模拟了单音干扰，宽带干扰以及部分频带干扰，输出误码率曲线以及各节点图像。
调制后的信号在特定的时间间隔内会迅速跳变到不同的频率，这些频率在预先确定的跳频表中随机选取。在仿真中，宽带干扰是通过一个覆盖整个频谱范围的噪声信号来实现的，而窄带干扰则是通过若干个覆盖部分频谱的单音信
阅读更多2024-11-26
【cocos creator】下拉框
【代码】【cocos creator】下拉框。
阅读更多2024-11-26
无人机飞行姿态详解！
例如，基于深度学习的姿态估计算法、基于视觉的避障算法等，这些新技术和新算法的应用将进一步推动无人机技术的发展和创新。这些技术和算法的应用可以显著提高无人机的飞行性能和稳定性，使其在各种复杂环境下都能够
阅读更多2024-11-26
commitlint校验git提交信息
commitlint校验git提交信息。
阅读更多2024-11-26
Python+Selenium无头浏览器实现网页截图
Selenium就是个自动化测试框架，底层还可以切换控制Chrome、火狐等等相关浏览器驱动。。可以使用Python+Selenium的方式进行网页截图等等相关工作
阅读更多2024-11-26
11.25 深度学习-pytorch模型组件
sgd=optim.SGD(model1.parameters(),lr=0.01) # model1.parameters()能够返回模型的参数（初始化的W和预测的值）传给optim.SGD
阅读更多2024-11-26
从0开始深度学习（31）——循环神经网络
前面介绍了n元语法模型，里面有一个叫隐状态，也被叫做隐藏变量，循环神经网络（recurrent neural networks，RNNs）是具有的神经网络。
阅读更多2024-11-26

【Spark】【大数据技术基础】课程 实验七 Spark基础编程实验

实验七：Spark初级编程实践

一、实验目的

二、实验平台

三、实验步骤

1. 准备工作（镜像已包含，不用安装）

2. Spark读取文件系统的数据

3. 编写独立应用程序实现数据去重

4. 编写独立应用程序实现求平均值问题

四、实验总结

相关文章

【Spark】【大数据技术基础】课程实验七 Spark基础编程实验