0


spark实验三-spark进阶编程

1.Spark编程统计各地区租房人数

实验目标:

(1) 掌握在IntelliJ IDEA 中操作spark程序开发

(2) 打包程序提交集群运行

实验说明:

现有一份某省份各地区租房信息文件 house.txt,文件中共有8个数据字段,字段说明如下表所示:

字段名称

说明

租房ID

租房编号

标题

发布的租房标题

链接

网址,可查看租房信息

地区

房子所在地区

地点

房子所在城市地点

地铁站

附近的地铁站

出租房数

可出租的房子数量

日期

发布日期

请在IntelliJ IDEA 中进行spark编程统计各地区的租房人数,完成编译后打包spark工程,通过spark-submit提交程序至集群中运行。

实现思路及步骤:

(1) 读取数据并创建RDD

val lines = sc.textFile("file:///root/doc/house.txt")

(2) 清洗数据,例如某些地铁站数据为空

val clean_lines = lines.filter(line => {
val fields = line.split("\s+") // 使用正则表达式分割字段
fields.length > 5 && fields(5).trim.nonEmpty // 确保至少有6个字段且第6个字段不为空
})

(3) 使用reduceByKey()方法统计人数

// 将清洗后的数据的第一列作为键,1作为值,转换为键值对
val categoryPairs = clean_lines.map(line => {
val fields = line.split("\s+")
(fields(0).trim, 1) // 假设第一列是分类信息
})

// 使用reduceByKey聚合相同分类的计数
val categoryCounts = categoryPairs.reduceByKey(_ + _)

categoryCounts.count()

(4) 使用saveAsTextFile()保存数据到hdfs

val hdfsOutputPath = "hdfs://master:9000/user/hadoop/clean_lines"
val lines = sc.textFile("file:///root/doc/house.txt")
lines.saveAsTextFile(hdfsOutputPath)
hdfs dfs -cat /user/hadoop/clean_lines/part-00000

2.自定义分区器事先按照人物标签进行数据分区

实验目标:

  1. 掌握使用spark自定义分区
  2. 掌握打包spark工程
  3. 掌握通过spark-submit提交应用

实验说明

先有一份某年度中国女排集训运动员数据文件 Volleyball_Plaryer.csv ,数据字段说明如下表所示

现要求在IntelliJ IDEA 中进行spark编程,通过自定义分区实现将运动员按照所属位置进行分区,并将程序打包,通过spark-submit提交应用。按照“主攻,接应,二传,副攻,自由人”五个标签设置五个分区,将分区结果输出到hdfs上。其中一个分区的结果举例如图所示:

实现思路及步骤:

  1. 使用textFile()方法读取数据创建RDD,并设置分区数为5
  2. 使用map()方法将数据输入数据按都好进行分割,筛选出position和name字段,并转化为(Positon,Name)的形式
  3. 自定义MyPartioner类,继承该类,重写类里面的numPartions和getPartition 方法。
  4. 在主函数中调用自定义分区类MyPartioner
  5. 打包spark工程,将应用程序提交至集群运行

如何在idea使用scala操作可以参考

IDEA使用SCALA-CSDN博客

import org.apache.spark.api.java.JavaRDD.fromRDD
import org.apache.spark.repl.Main.conf
import org.apache.spark.{Partitioner, SparkConf, SparkContext}

object Main {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf()
    sparkConf.setMaster("local")   //本地单线程运行
    sparkConf.setAppName("Main")
    val sc = new SparkContext(sparkConf)
    val lines = sc.textFile("/root/doc/Volleyball_Players.csv",5)
    val  new_lines= lines.map(line => {
      val fields = line.split(",")
      val position = fields(5).trim
      val name = fields(0).trim
      (position, name)
    })
    val myPartitioner: MyPartitioner = new MyPartitioner(5)
    new_lines.repartition(myPartitioner.numPartitions)
    new_lines.foreach(println)
  }
}
class MyPartitioner(override val numPartitions: Int) extends Partitioner {
  override def getPartition(key: Any): Int = key match {
    case (position: String) => position.hashCode % numPartitions
  }
}

打包代码

打包成功后在你本机找到该文件

传到linux下

将Volleyball_Players.csv上传到hdfs目录下

hdfs dfs -mkdir -p /root/doc
hdfs dfs -put Volleyball_Players.csv /root/doc/

此时将代码提交
spark-submit --master yarn --class Main untitled4.jar

--master yarn

指定了 Spark 应用程序应该运行在 YARN(Yet Another Resource Negotiator)集群管理器上。

--class Main

指示 Spark 应用程序的入口点是

Main

类中的

main

方法。您需要将

Main

替换为您实际的主类名。

untitled4.jar

是包含您 Spark 应用程序的 JAR 文件。


本文转载自: https://blog.csdn.net/qq_58503141/article/details/137771218
版权归原作者 浮萍哥 所有, 如有侵权,请联系我们删除。

“spark实验三-spark进阶编程”的评论:

还没有评论