使用Scala语言编写Spark应用程序实现数据去重_scala去重

一、题目需求

使用Scala语言编写Spark应用程序实现数据去重（需要使用编译打包工具Maven或sbt进行编译打包）。
对于两个输入文件A和B，编写Spark独立应用程序，对两个文件进行合并，并剔除其中重复的内容，得到一个新文件C。下面是输入文件和输出文件的一个样例，供参考。
输入文件A的样例如下：
20180901 x
20180902 y
20180903 x
20180904 y
20180905 z
20180906 z
输入文件B的样例如下：
20180901 y
20180902 y
20180903 x
20180904 z
20180905 y
根据输入的文件A和B合并得到的输出文件C的样例如下：
20180901 x
20180901 y
20180902 y
20180903 x
20180904 y
20180904 z
20180905 y
20180905 z
20180906 z

二、建立目录结构

• 可以在根目录文件夹下以此创建目录及文件，形成最后的目录树
  

  （一）创建 sparkapp4 文件夹并切换

  mkdir sparkapp4 
  1

  cd ~/sparkapp4
  1

  （二）创建 data 文件夹（存放A.txt B.txt）

  mkdir data
  1

  将题目要求去重的两个文本文件放到data文件夹下

  vim data/A.txt
  1

  vim data/B.txt
  1

  （三）创建 /src/main/scala 文件夹用来存放代码

  mkdir -p src/main/scala
  1

  （四）编写代码

  vim src/main/scala/De_duplication.scala
  1

  import org.apache.spark.SparkContext
  import org.apache.spark.SparkContext._
  import org.apache.spark.SparkConf
  import org.apache.spark.HashPartitioner
  object De_duplication {
          def main(args: Array[String]) {
              val conf = new SparkConf().setAppName("De_duplication")
              val sc = new SparkContext(conf)
              val dataFile = "file:///home/hadoop/sparkapp4/data"
              val data  = sc.textFile(dataFile,2)
              val res = data.filter(_.trim().length>0).map(line=>(line.trim,"")).partitionBy(new HashPartitioner(1)).groupByKey().sortByKey().keys
              res.collect().foreach(println) //这句是运行是为了用grep筛选结果，可删，不影响结果
              res.saveAsTextFile("file:///home/hadoop/sparkapp4/result")
    }
  }
  
  1
  2
  3
  4
  5
  6
  7
  8
  9
  10
  11
  12
  13
  14
  15
  16

  这里需要注意的点有：

  • 文件名涉及到最后运行时的内容，需要注意
  • datafile 的绝对路径名为你存储A.txt、B.txt两个文件的文件夹名文件夹下仅放要去重的文件
  • 最后存储路径同理，这里我们最后的结果存储在 ~/sparkapp4/result的文件夹下，文件名为默认的 part-00000.txt

  （五）编写 simple.sbt

  vim simple.sbt
  1

  name := "Simple Project"
  version := "1.0"
  scalaVersion := "2.11.8"
  libraryDependencies += "org.apache.spark" %% "spark-core" % "2.1.0"
  
  1
  2
  3
  4
  5

  注意scalaversion和libraryDependencies中的版本号，不清楚的可以运行一下spark-shell查看
  

  cd /usr/local/spark
  1

  /bin/spark-shell
  1

最后检查目录树上面的结构即可

三、编译及运行

（一）编译

在 sparkapp4 文件夹下

/usr/local/sbt/sbt package
1

（二）运行

	/usr/local/spark/bin/spark-submit --class "De_duplication" ~/sparkapp4/target/scala-2.11/simple-project_2.11-1.0.jar 2>&1 | grep "2018"
1

注意：

• "De_duplication" 为代码的object名（同文件名）
• ~/sparkapp4/target/scala-2.11/simple-project_2.11-1.0.jar为编译后生成的jar文件，可以自行到相关目录下查找具体的文件名并更改
• grep “2018” 过滤结果

（三）验证

结果正确！