当前位置：网站首页>spark on yarn

spark on yarn

2022-08-02 14:05:00 【boyzwz】

提交代码到yarn模式运行

准备代码：

package com.shujia.spark.core

import com.shujia.spark.util.HdfsUtil
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object HomeWork2 {
  def main(args: Array[String]): Unit = {
    /**
     * 创建spark环境
     */
    val conf = new SparkConf()
    conf.setAppName("HomeWork2")
    /**
     * 提交代码到yarn模式运行，需将local本地模式注释
     */
    //    conf.setMaster("local")
    val sc = new SparkContext(conf)

    /**
     * 2、统计总分大于年级平均分的学生
     */
    //读取hdfs上的文件
    val linesRDD: RDD[String] = sc.textFile("/spark_data/score.txt")
    //获取学生id和分数
    val scoreRDD: RDD[(String, Int)] = linesRDD
      //切分
      .map(lines => lines.split(","))
      //过滤脏数据
      .filter(arr => arr.length == 3)
      //取数据
      .map(arr => (arr(0), arr(2).toInt))
    //求出每个学生总分
    val sumRDD: RDD[(String, Int)] = scoreRDD.reduceByKey((x, y) => x + y)
    //求出年级平均分
    val scores: RDD[Int] = sumRDD.map(kv => kv._2)
    val avg_score: Double = scores.sum() / scores.count()
    //过滤出高于平均分的学生
    val bigRDD: RDD[(String, Int)] = sumRDD.filter(kv => kv._2 > avg_score)
    //处理数据
    val resultRDD: RDD[String] = bigRDD.map(kv => s"${kv._1}\t${kv._2}")
    //删除路径；自定义封装类，若hdfs上路径已存在，则删除路径
    HdfsUtil.delete("/spark_data/big_avg")
    //保存数据到hdfs上
    resultRDD.saveAsTextFile("/spark_data/big_avg")
  }

}