Spark2.4.8 求員工工資總額

一、測驗資料

• 員工資訊表：下載地址
• 表欄位說明：
  

二、任務要求

• 任務1：求每個部門的員工工資總額
• 任務2：求每個部門員工工資與獎金總額
• 任務3：將任務2中的結果按照部門號進行升序排
• 任務4：將任務2中的結果按照工資總額進行降序排

三、編碼實作

• 創建maven工程

• 添加spark相關依賴，在pom.xml中添加如下依賴

  <packaging>jar</packaging>
  
  <properties>
      <scala.version>2.11.8</scala.version>
      <spark.version>2.4.8</spark.version>
      <spark.artifact.version>2.12</spark.artifact.version>
      <hadoop.version>2.7.3</hadoop.version>
  </properties>
  
  <dependencies>
      <dependency>
          <groupId>org.apache.spark</groupId>
          <artifactId>spark-core_${spark.artifact.version}</artifactId>
          <version>${spark.version}</version>
      </dependency>
      <dependency>
          <groupId>org.apache.hadoop</groupId>
          <artifactId>hadoop-client</artifactId>
          <version>${hadoop.version}</version>
      </dependency>
      <!-- 使用scala2.11.8進行編譯和打包 -->
      <dependency>
          <groupId>org.scala-lang</groupId>
          <artifactId>scala-library</artifactId>
          <version>${scala.version}</version>
      </dependency>
  </dependencies>
  
  <build>
      <!-- 指定scala源代碼所在的目錄 -->
      <sourceDirectory>src/main/scala</sourceDirectory>
      <testSourceDirectory>src/test/scala</testSourceDirectory>
  
      <plugins>
          <!--對src/main/java下的后綴名為.java的文件進行編譯 -->
          <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-compiler-plugin</artifactId>
              <version>3.8.1</version>
              <configuration>
                  <source>1.8</source>
                  <target>1.8</target>
              </configuration>
          </plugin>
          <plugin>
              <groupId>org.apache.maven.plugins</groupId>
              <artifactId>maven-assembly-plugin</artifactId>
              <configuration>
                  <descriptorRefs>
                      <descriptorRef>jar-with-dependencies</descriptorRef>
                  </descriptorRefs>
              </configuration>
          </plugin>
          <!-- scala的打包插件 -->
          <plugin>
              <groupId>net.alchim31.maven</groupId>
              <artifactId>scala-maven-plugin</artifactId>
              <version>4.5.4</version>
              <executions>
                  <execution>
                      <goals>
                          <goal>compile</goal>
                          <goal>testCompile</goal>
                      </goals>
                  </execution>
              </executions>
              <configuration>
                  <scalaVersion>${scala.version}</scalaVersion>
              </configuration>
          </plugin>
      </plugins>
  </build>
  

• 【任務一】代碼實作如下：

  • 實作代碼：

    import org.apache.spark.{SparkConf, SparkContext}
    
    object CountSalary {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName(CountSalary.getClass.getName).setMaster("local[2]")
        val sc = new SparkContext(sparkConf)
        sc.textFile("d:/Tools/emp.csv")
          .map(line => {
            val strings = line.split(",")
            val salary = strings(5).toInt
            val deptNo = strings(7).toInt
            (deptNo,salary)
          })
          .reduceByKey(_+_)
          .collect()
          .foreach(println)
        // 關閉sc
        sc.stop()
      }
    }
    

  • 結果：
    

• 【任務二】代碼實作如下：

  • 實作代碼：

    import org.apache.spark.{SparkConf, SparkContext}
    
    object CountBonusAndSalary {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName(CountBonusAndSalary.getClass.getName).setMaster("local[2]")
        val sc = new SparkContext(sparkConf)
        sc.textFile("d:/Tools/emp.csv")
          .map(line => {
            val strings = line.split(",")
            val salary = strings(5).toInt
            val deptNo = strings(7).toInt
            var bonus = 0
            if (!"".equals(strings(6)) && null != strings(6)){
              bonus = strings(6).toInt
            }
            (deptNo,salary+bonus)
          })
          .reduceByKey(_+_)
          .collect()
          .foreach(println)
        // 關閉sc
        sc.stop()
      }
    }
    

  • 結果如下：
    

• 【任務三】代碼實作如下：

  • 代碼實作

    import org.apache.spark.{SparkConf, SparkContext}
    
    object CountBonusAndSalaryByAsc {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName(CountBonusAndSalaryByAsc.getClass.getName).setMaster("local[2]")
        val sc = new SparkContext(sparkConf)
        sc.textFile("d:/Tools/emp.csv")
          .map(line => {
            val strings = line.split(",")
            val salary = strings(5).toInt
            val deptNo = strings(7).toInt
            var bonus = 0
            if (!"".equals(strings(6)) && null != strings(6)){
              bonus = strings(6).toInt
            }
            (deptNo,salary+bonus)
          })
          .reduceByKey(_+_)
          .sortByKey(true)
          .collect()
          .foreach(println)
        // 關閉sc
        sc.stop()
      }
    }
    
    

  • 結果如下
    

• 【任務四】代碼實作如下：

  • 代碼實作：

    import org.apache.spark.{SparkConf, SparkContext}
    
    object CountTotalByAsc {
      def main(args: Array[String]): Unit = {
        val sparkConf = new SparkConf().setAppName(CountTotalByAsc.getClass.getName).setMaster("local[2]")
        val sc = new SparkContext(sparkConf)
        sc.textFile("d:/Tools/emp.csv")
          .map(line => {
            val strings = line.split(",")
            val salary = strings(5).toInt
            val deptNo = strings(7).toInt
            var bonus = 0
            if (!"".equals(strings(6)) && null != strings(6)){
              bonus = strings(6).toInt
            }
            (deptNo,salary+bonus)
          })
          .reduceByKey(_+_)
          .sortBy(tuple2 => {
            tuple2._2
          },false)
          .collect()
          .foreach(println)
        // 關閉sc
        sc.stop()
      }
    }
    
    

  • 結果如下：
    

四、實驗要求

• 請使用不同的算子來完成如上的任務