eclipse - 如何比较 2 列并在 Scala 中连接

Question

这是我的文本文件，它是程序的输入:

Id       Title Copy
B2002010 gyh   1
D2001001 abc   12
M2003005 zxc   3
D2002003 qwe   13
M2001002 efg   1
D2001004 asd   6
D2003005 zxc   3
M2001006 wer   6
D2001006 wer   6
B2004008 sxc   10
D2002007 sdf   9
D2004008 sxc   10

ID 的格式为 Xyyyyrrr，其中:

• X 是 B => 书或 M => 杂志
• yyyy 是年份
• rrr为随机数

我要做的是:获取同一年的书籍或杂志的总份数。另外，对“复制”列进行了一次小数据清理，如果我发现数字以外的内容，我会将其替换为“0”。

我的 Spark 项目在 Eclipse 上，我正在使用 Maven 和 Scala IDE我需要使用 MapReduce 函数。

我已经启动了拆分文本文件的 Map 函数。

这是我开始的代码:

package bd.spark_app

import org.apache.spark.SparkConf
import org.apache.spark.SparkContext
import org.apache.spark.SparkContext._
import org.apache.spark.sql.SQLContext
import org.apache.spark.sql._
import org.apache.spark.sql.types.IntegerType
import org.apache.spark.rdd.RDD.rddToOrderedRDDFunctions
import scala.io.Source
import org.apache.spark.sql.functions._
import scala.collection.mutable.WrappedArray
import org.apache.log4j._
import org.apache.spark.sql.types.{StructType, StructField, StringType}
import org.apache.spark.sql.Row
import scala.Array

object alla {
  def main(args:Array[String]) = {
    val conf = new SparkConf().setMaster("local").setAppName("trying")
    val sc = new SparkContext(conf)
    val x = sc.textFile("/home/hadoopusr/sampledata")

    x.map(_.split(" ")).foreach(r => 
      println(r(0).dropRight(3), r(2))
    )

    sc.stop()
  }
} 

这是我上面显示的 Map 函数的结果

(B2002,1)
(D2001,12)
(M2003,3)
(D2002,13)
(M2001,1)
(D2001,6)
(D2003,3)
(M2001,6)
(D2001,6)
(B2004,10)
(D2002,9)
(D2004,10)
(M2004,11)
(D2004,11)

我只需要某种 reduce 函数，它可以抓取同一年的所有书籍和杂志，并将副本数加在一起，并检查“副本”列是否为数字

示例:对于记录 (B2002,12) 和 (B2002,16)，结果应为 (B2002,28)。

Answer 1

可以使用方法“reduceByKey”:

val converted = x.map(_.split(" ")).map(r => (r(0).dropRight(3), r(2).toInt))
val result = converted.reduceByKey(_ + _)

输出:

(M2001,7)
(D2001,24)
(M2003,3)
(D2003,3)
(D2002,22)
(D2004,10)
(B2002,1)
(B2004,10)

注意: 看起来输入文件是“csv”格式，最好使用“spark.read.csv”读取数据，并使用 DataFrame 而不是 RDD。