当前位置：首页 > news >正文

spark(day02)

news 来源：原创 2024/5/5 13:19:09

案例
求中位数

package cn.tedu.wordcount

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object DriverMedian {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("median")
    val sc=new SparkContext(conf)
    val data=sc.textFile("D://未来/spark练习文件/median.txt", 2)
    val r1=data.flatMap { line => line.split(" ") }.map { x => x.toInt }.sortBy{x=>x}
    val num=r1.count().toInt
    if(num%2==0){
      val m=r1.take(num/2+2)
      val n=r1.take(num/2)
      println((m.sum-n.sum)/2.toDouble)
    }else{
      println(r1.take((num+1)/2).last.toDouble)
    }
    
  }
}

二次计数

package cn.tedu.wordcount

import org.apache.spark.SparkContext
import org.apache.spark.SparkConf

object DriverSsort {
  def main(args: Array[String]): Unit = {
    val conf=new SparkConf().setMaster("local").setAppName("median")
    val sc=new SparkContext(conf)
    val data=sc.textFile("D://未来/spark练习文件/ssort.txt", 1)
    
    // --RDD[String:line]->RDD[ (Ssort,line)]->SortByKey

    val r1=data.map { line => 
      val info=line.split(" ")
      val col1=info(0)
      val col2=info(1).toInt
      (new Ssort(col1,col2),line) 
    }
    //true----按类里定义的排序规则排序
    val r2=r1.sortByKey(true).map{x=>x._2}
    r2.foreach(println)
  }
}

package cn.tedu.wordcount

import java.io.Serializable
/* 
 * ssort中的两列数据
 * 引入分类方法 ctrl+1
 * */
class Ssort(val col1:String, val col2:Int) extends Ordered[Ssort] with Serializable{
  /*
	*此方法用于指定排序规则
	* 目标:先按第一列做升序排序
	* 再按第二列做降序排序
	*/

  def compare(that: Ssort): Int = {
    //--获取第一列升序排序比较的结果
    val result=this.col1.compareTo(that.col1)
    if(result==0){
      //--第一列升序排完后，按第二列做降序
      that.col2.compare(this.col2)
    }else{
      result
    }
  }
}

spark环境执行代码

package cn.tedu.wordcount2

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext

object Driver {
  def main(args: Array[String]): Unit = {
    //表示将代码运行在Spark集群环境下
    val conf=new SparkConf().setMaster("spark://hadoop01:7077").setAppName("wordcount2")
    val sc=new SparkContext(conf)
    val data=sc.textFile("hdfs://hadoop01:9000/data",3)
    
    val result=data.flatMap { x => x.split(" ") }.map { x => (x,1) }.reduceByKey{_+_}
    
    result.saveAsTextFile("hdfs://hadoop01:9000/result06")
  }
}

步骤
1.打成jar包
在这里插入图片描述
2.把jar包放到bin目录，执行指令，hafs下生成数据

[root@hadoop01 bin]# pwd
/home/presoftware/spark-2.0.1-bin-hadoop2.7/bin


[root@hadoop01 bin]# ll wor*
-rw-r--r--. 1 root root 4578 9月   3 2022 wordcount.jar

#提交jar包指令---执行代码
[root@hadoop01 bin]# sh spark-submit --class cn.tedu.wordcount2.Driver wordcount.jar