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 文件。