6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

目录

3.6 Action

3.7 Key-Value RDD操作

3.7.1 创建Pair RDD

3.7.2 Transformation操作

          1、类似 map 操作

          2、聚合操作【重要、难点】

          3、排序操作

          4、join操作

3.7.3 Action操作

3.8 输入与输出

3.8.1 文件输入与输出

          1、文本文件

          2、csv文件

          3、json文件

          4、SequenceFile

          5、对象文件

3.8.2 JDBC

3.9 算子综合应用案例

1、WordCount - scala

2、WordCount - java

3、计算圆周率

4、广告数据统计

5、找共同好友

6、Super WordCount


 

3.6 Action

Action 用来触发RDD的计算,得到相关计算结果;

Action触发Job。一个Spark程序(Driver程序)包含了多少 Action 算子,那么就有多少Job;
典型的Action算子: collect / count
collect() => sc.runJob() => ... => dagScheduler.runJob() => 触发了Job
要求:能快速准确的区分Transformation、Action

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

collect() / collectAsMap()
stats / count / mean / stdev / max / min
reduce(func) / fold(func) / aggregate(func)

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

first():Return the first element in this RDD
take(n):Take the first num elements of the RDD
top(n):按照默认(降序)或者指定的排序规则,返回前num个元素。
takeSample(withReplacement, num, [seed]):返回采样的数据
foreach(func) / foreachPartition(func):与map、mapPartitions类似,区别是 foreach 是 Action
saveAsTextFile(path) / saveAsSequenceFile(path) / saveAsObjectFile(path)

// 返回统计信息。仅能作用 RDD[Double] 类型上调用
val rdd1 = sc.range(1, 101)
rdd1.stats    // 查看rdd1的统计信息, 其中项目可以单独调用
// res14: org.apache.spark.util.StatCounter = (count: 100, mean: 50.500000, stdev: 28.866070, max: 100.000000, min: 1.000000)

val rdd2 = sc.range(1, 101)
// 不能调用
rdd1.zip(rdd2).stats  // 多组元素用不了
// <console>:28: error: value stats is not a member of org.apache.spark.rdd.RDD[(Long, Long)]

// count在各种类型的RDD上,均能调用
rdd1.zip(rdd2).count
// res16: Long = 100

// 聚合操作
val rdd = sc.makeRDD(1 to 10, 2)
rdd.reduce(_+_)
// res17: Int = 55

// fold和reduce的区别就是, 可以给一个初值
rdd.fold(0)(_+_)    // 可以给初值0
// res18: Int = 55

rdd.fold(1)(_+_)    // 给初值1, 结果多3, 详细解释看下面的图
// res19: Int = 58

rdd.fold(1)((x, y) => {
    println(s"x=$x, y=$y")
    x+y
})
// 详细结果, 参见下面的图 与a+b 和 x+y 一样

rdd.aggregate(0)(_+_, _+_)  // 初始值的数据类型要一致才行
// res7: Int = 55

rdd.aggregate(1)(_+_, _+_)    // _+_ 局部汇总规则  _+_ 全局汇总规则
// res8: Int = 58

// rdd.getNumPartitions
// res18: Int = 2  分区是2, 所以下面 +3
rdd.aggregate(1)(
    (a, b) => {
        println(s"a=$a, b=$b")
        a+b    
    },
    (x, y) => {
        println(s"x=$x, y=$y")
        x+y
    }
)
// 详细结果, 参见下面的图


====================================================
// first / take(n) / top(n) :获取RDD中的元素。多用于测试
rdd.first
// res10: Int = 1

rdd.take(10)    // 返回前10个元素
// res11: Array[Int] = Array(1, 2, 3, 4, 5, 6, 7, 8, 9, 10)

rdd.top(10)    // 返回最大的前10个
// res12: Array[Int] = Array(10, 9, 8, 7, 6, 5, 4, 3, 2, 1)

// 采样并返回结果
rdd.takeSample(false, 5)
// res13: Array[Int] = Array(9, 2, 4, 8, 1)

// 保存文件到指定路径(rdd有多少分区,就保存为多少文件)
// 注意小文件问题, 可以使用coalesce缩减分区
rdd.coalesce(1).saveAsTextFile("data/t1")


val rdd = (1 to 10)
rdd: scala.collection.immutable.Range.Inclusive = Range 1 to 10
// foreach 没有返回值, 需要借助其他函数, 把结果发到其他地方
rdd.foreach(x => x + "")
// 无返回值
rdd.foreach(x => print(x + " "))    
// 1 2 3 4 5 6 7 8 9 10 

 

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

在使用 fold(1) 之后, 结果增加了3的原因:  有几个分区就会局部汇总几次, 另外还有全局汇总,  每次汇总都会 + 初始值

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

 

3.7 Key-Value RDD操作

RDD整体上分为 Value 类型和 Key-Value 类型。
前面介绍的是 Value 类型的RDD的操作,实际使用更多的是 key-value 类型的RDD,也称为 PairRDD。
Value 类型RDD的操作基本集中在 RDD.scala 中;
key-value 类型的RDD操作集中在 PairRDDFunctions.scala 中;

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

前面介绍的大多数算子对 Pair RDD 都是有效的。Pair RDD还有属于自己的 Transformation、Action 算子;

 

3.7.1 创建Pair RDD

val arr = (1 to 10).toArray
val arr1 = arr.map(x => (x, x*10, x*100))

// rdd1 不是 Pair RDD
val rdd1 = sc.makeRDD(arr1)
rdd1.take(3)
// res1: Array[(Int, Int, Int)] = Array((1,10,100), (2,20,200), (3,30,300))        

// rdd2 是 Pair RDD
val arr2 = arr.map(x => (x, (x*10, x*100)))

val rdd2 = sc.makeRDD(arr2)
rdd2.take(3)
// res2: Array[(Int, (Int, Int))] = Array((1,(10,100)), (2,(20,200)), (3,(30,300)))


rdd1.map(x => (x._1, (x._2, x._3))).collectAsMap  // 这种方式可以转换成map
// res5: scala.collection.Map[Int,(Int, Int)] = Map(8 -> (80,800), 2 -> (20,200), 5 -> (50,500), 4 -> (40,400), 7 -> (70,700), 10 -> (100,1000), 1 -> (10,100), 9 -> (90,900), 3 -> (30,300), 6 -> (60,600))

 

3.7.2 Transformation操作

          1、类似 map 操作

val a = sc.parallelize(List((1,2),(3,4),(5,6)))
a.collect
// res6: Array[(Int, Int)] = Array((1,2), (3,4), (5,6))

// 使用 mapValues 更简洁
val b = a.mapValues(x=>1 to x)  // 对每个value执行, value => 1 to value
b.collect
// Array((1,Range 1 to 2), (3,Range 1 to 4), (5,Range 1 to 6))

// 可使用map实现同样的操作
val c = a.map(x => (x._1, 1 to x._2))
c.collect
// Array((1,Range 1 to 2), (3,Range 1 to 4), (5,Range 1 to 6))
val c = a.map{case (k, v) => (k, 1 to v)}    // 偏函数方式
c.collect
// Array((1,Range 1 to 2), (3,Range 1 to 4), (5,Range 1 to 6))

// flatMapValues 将 value 的值压平
val c = a.flatMapValues(x=>1 to x)
c.collect
// res10: Array[(Int, Int)] = Array((1,1), (1,2), (3,1), (3,2), (3,3), (3,4), (5,1), (5,2), (5,3), (5,4), (5,5), (5,6))

val c = a.mapValues(x=>1 to x)
// Array((1,Range 1 to 2), (3,Range 1 to 4), (5,Range 1 to 6))
val c = a.mapValues(x=>1 to x).flatMap{case (k, v) => v.map(x => (k, x))}  // 使用偏函数, 计算 v. 丢掉 k
c.collect
// res10: Array[(Int, Int)] = Array((1,1), (1,2), (3,1), (3,2), (3,3), (3,4), (5,1), (5,2), (5,3), (5,4), (5,5), (5,6))

c.keys
// res11: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[8] at keys at <console>:26

c.values
// res12: org.apache.spark.rdd.RDD[Int] = MapPartitionsRDD[9] at values at <console>:26

c.map{case (k, v) => k}.collect  // k留下, v丢掉
// res16: Array[Int] = Array(1, 1, 3, 3, 3, 3, 5, 5, 5, 5, 5, 5)

c.map{case (k, _) => k}.collect  // k留下, 其他丢掉
// res17: Array[Int] = Array(1, 1, 3, 3, 3, 3, 5, 5, 5, 5, 5, 5)

c.map{case (_, v) => v}.collect  // v留下, k丢掉
// res18: Array[Int] = Array(1, 2, 1, 2, 3, 4, 1, 2, 3, 4, 5, 6)

 

          2、聚合操作【重要、难点】

PariRDD(k, v) 使用范围广,聚合
groupByKey / reduceByKey / foldByKey / aggregateByKey
combineByKey(OLD) / combineByKeyWithClassTag (NEW) => 底层实现, 版本区别
subtractByKey:类似于subtract,删掉 RDD 中键与 other RDD 中的键相同的元素

 

小案例:给定一组数据:("spark", 12), ("hadoop", 26), ("hadoop", 23), ("spark", 15), ("scala", 26), ("spark", 25),("spark", 23), ("hadoop", 16), ("scala", 24), ("spark", 16), 键值对的key表示图书名称,value表示某天图书销量。计算每个键对应的平均值,也就是计算每种图书的每天平均销量。

val rdd = sc.makeRDD(Array(("spark", 12), ("hadoop", 26), ("hadoop", 23), ("spark", 15),
("scala", 26), ("spark", 25), ("spark", 23), ("hadoop", 16), ("scala", 24), ("spark",16)))

rdd.groupByKey().collect        // 下面是scala的数据结构
// Array((scala,CompactBuffer(26, 24)), (hadoop,CompactBuffer(26, 23, 16)), (spark,CompactBuffer(12, 25, 15, 16, 23)))

// groupByKey
rdd.groupByKey().map(x=>(x._1, x._2.sum.toDouble/x._2.size)).collect    // groupByKey不加括号也能执行
// Array((scala,25.0), (hadoop,21.666666666666668), (spark,18.2))
rdd.groupByKey().map{case (k, v) => (k, v.sum * 1.0/v.size)}.collect    // 可读性更强
// 同上
rdd.groupByKey.mapValues(v => v.sum.toDouble/v.size).collect
// 同上

// reduceByKey
rdd.mapValues((_, 1)).                            // Array((spark,(12,1)), (hadoop,(26,1)) ... )
reduceByKey((x, y)=> (x._1+y._1, x._2+y._2)).     // Array((scala,(50,2)), (hadoop,(65,3)), (spark,(91,5)))
mapValues(x => (x._1.toDouble / x._2)).
collect()
// Array((scala,25.0), (hadoop,21.666666666666668), (spark,18.2))

// foldByKey, 与上面类似, 最大的区别是可以定义初值
rdd.mapValues((_, 1)).foldByKey((0, 0))((x, y) => {
    (x._1+y._1, x._2+y._2)
}).mapValues(x=>x._1.toDouble/x._2).collect



// aggregateByKey
// aggregateByKey => 定义初值 + 分区内的聚合函数 + 分区间的聚合函数
rdd.mapValues((_, 1)).aggregateByKey((0,0))(
    (x, y) => (x._1 + y._1, x._2 + y._2),
    (a, b) => (a._1 + b._1, a._2 + b._2)
).mapValues(x=>x._1.toDouble / x._2).collect

// 初值(元组)与RDD元素类型(Int)可以不一致
rdd.aggregateByKey((0, 0))(
    (x, y) => {println(s"x=$x, y=$y"); (x._1 + y, x._2 + 1)},        // x,y分别是前后的一个数, 这里的x类型被限定为 (_1, _2), 于是y传进来被加到 x._1上, x._2自增加1
    (a, b) => {println(s"a=$a, b=$b"); (a._1 + b._1, a._2 + b._2)}   // 这里的 a,b 分别是前后数, 格式都是 (_1, _2)
).mapValues(x=>x._1.toDouble/x._2).collect

// 分区内的合并与分区间的合并,可以采用不同的方式;这种方式是低效的!
rdd.aggregateByKey(scala.collection.mutable.ArrayBuffer[Int]())(    // 需要导包
    (x, y) => {x.append(y); x},    // 这里 x 要返回
    (a, b) => {a++b}
).mapValues(v => v.sum.toDouble/v.size).collect

// combineByKey(理解就行)
rdd.combineByKey(
    (x: Int) => {println(s"x=$x"); (x,1)},
    (x: (Int, Int), y: Int) => {println(s"x=$x, y=$y");(x._1+y, x._2+1)},
    (a: (Int, Int), b: (Int, Int)) => {println(s"a=$a, b=$b");(a._1+b._1, a._2+b._2)}
).mapValues(x=>x._1.toDouble/x._2).collect

// subtractByKey
val rdd1 = sc.makeRDD(Array(("spark", 12), ("hadoop", 26), ("hadoop", 23), ("spark",15)))
val rdd2 = sc.makeRDD(Array(("spark", 100), ("hadoop", 300)))
rdd1.subtractByKey(rdd2).collect()
// 相同key消失

// subtractByKey
val rdd = sc.makeRDD(Array(("a",1), ("b",2), ("c",3), ("a",5), ("d",5)))
val other = sc.makeRDD(Array(("a",10), ("b",20), ("c",30)))
rdd.subtractByKey(other).collect()
// 相同k消失, 和 v 无关

 

结论:效率相等用最熟悉的方法;groupByKey在一般情况下效率低,尽量少用 (95%以上不用)
初学:最重要的是实现;如果使用了groupByKey,寻找替换的算子实现;

 

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

groupByKey Shuffle过程中传输的数据量大,效率低

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

          3、排序操作

sortByKey:sortByKey函数作用于PairRDD,对Key进行排序。在org.apache.spark.rdd.OrderedRDDFunctions 中实现:

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

val a = sc.parallelize(List("wyp", "iteblog", "com", "397090770", "test"))
val b = sc.parallelize (1 to a.count.toInt)
val c = a.zip(b)

// 默认使用升序
c.sortByKey().collect
// Array((397090770,4), (com,3), (iteblog,2), (test,5), (wyp,1))

// 使用降序
c.sortByKey(false).collect
// Array((wyp,1), (test,5), (iteblog,2), (com,3), (397090770,4))

 

          4、join操作

cogroup / join / leftOuterJoin / rightOuterJoin / fullOuterJoin

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

val rdd1 = sc.makeRDD(Array((1,"Spark"), (2,"Hadoop"), (3,"Kylin"), (4,"Flink")))
val rdd2 = sc.makeRDD(Array((3,"李四"), (4,"王五"), (5,"赵六"), (6,"冯七")))
val rdd3 = rdd1.cogroup(rdd2)
rdd3.collect.foreach(println)
// (6,(CompactBuffer(),CompactBuffer(冯七)))
// (1,(CompactBuffer(Spark),CompactBuffer()))
// (2,(CompactBuffer(Hadoop),CompactBuffer()))
// (3,(CompactBuffer(Kylin),CompactBuffer(李四)))
// (4,(CompactBuffer(Flink),CompactBuffer(王五)))
// (5,(CompactBuffer(),CompactBuffer(赵六)))

rdd3.filter{case (_, (v1, v2)) => v1.nonEmpty & v2.nonEmpty}.collect
// Array((3,(CompactBuffer(Kylin),CompactBuffer(李四))), (4,(CompactBuffer(Flink),CompactBuffer(王五))))

// 仿照源码实现join操作
rdd3.flatMapValues( pair =>
    for (v <- pair._1.iterator; w <- pair._2.iterator) yield (v, w)
).collect
// Array((3,(Kylin,李四)), (4,(Flink,王五)))

val rdd1 = sc.makeRDD(Array(("1","Spark"),("2","Hadoop"),("3","Scala"),("4","Java")))
val rdd2 = sc.makeRDD(Array(("3","20K"),("4","18K"),("5","25K"),("6","10K")))
rdd1.join(rdd2).collect
// Array((3,(Scala,20K)), (4,(Java,18K)))
rdd1.leftOuterJoin(rdd2).collect
// Array((1,(Spark,None)), (2,(Hadoop,None)), (3,(Scala,Some(20K))), (4,(Java,Some(18K))))

rdd1.rightOuterJoin(rdd2).collect
// Array((6,(None,10K)), (3,(Some(Scala),20K)), (4,(Some(Java),18K)), (5,(None,25K)))

rdd1.fullOuterJoin(rdd2).collect
// Array((6,(None,Some(10K))), (1,(Some(Spark),None)), (2,(Some(Hadoop),None)), (3,(Some(Scala),Some(20K))), (4,(Some(Java),Some(18K))), (5,(None,Some(25K))))

 

3.7.3 Action操作

collectAsMap / countByKey / lookup(key)
 

countByKey源码:

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

lookup(key):高效的查找方法,只查找对应分区的数据(如果该RDD有分区器的话)

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

val rdd1 = sc.makeRDD(Array(("1","Spark"),("2","Hadoop"),("3","Scala"),("1","Java")))
val rdd2 = sc.makeRDD(Array(("3","20K"),("4","18K"),("5","25K"),("6","10K")))
rdd1.lookup("1")
// res25: Seq[String] = WrappedArray(Spark, Java)

rdd2.lookup("3")
// res26: Seq[String] = WrappedArray(20K)

 

3.8 输入与输出

3.8.1 文件输入与输出

          1、文本文件

数据读取:textFile(String)。可指定单个文件,支持通配符。
这样对于大量的小文件读取效率并不高,应该使用 wholeTextFiles
def wholeTextFiles(path: String, minPartitions: Int = defaultMinPartitions): RDD[(String, String)])
返回值RDD[(String, String)],其中 Key是文件的名称,Value是文件的内容

 

数据保存:saveAsTextFile(String)。指定的输出目录。

 

          2、csv文件

读取 CSV(Comma-Separated Values)/TSV(Tab-Separated Values) 数据和读取 JSON 数据相似,都需要先把文件当作普通文本文件来读取数据,然后通过将每一行进行解析实现对CSV的读取。
CSV/TSV 数据的输出也是需要将结构化RDD通过相关的库转换成字符串RDD,然后使用 Spark 的文本文件 API 写出去。

 

          3、json文件

如果 JSON 文件中每一行就是一个JSON记录,那么可以通过将JSON文件当做文本文件来读取,然后利用相关的JSON库对每一条数据进行JSON解析。

JSON数据的输出主要是通过在输出之前将由结构化数据组成的 RDD 转为字符串 RDD,然后使用 Spark 的文本文件API 写出去。

json文件的处理使用SparkSQL最为简洁。

 

          4、SequenceFile

SequenceFile文件是Hadoop用来存储二进制形式的key-value对而设计的一种平面文件(Flat File)。 Spark 有专门用来读取 SequenceFile 的接口。在 SparkContext 中,可以调用:sequenceFile[keyClass, valueClass];

调用 saveAsSequenceFile(path) 保存PairRDD,系统将键和值能够自动转为Writable类型。

 

          5、对象文件

对象文件是将对象序列化后保存的文件,采用Java的序列化机制。

通过 objectFile[k,v](path) 接收一个路径,读取对象文件,返回对应的 RDD,也可以通过调用saveAsObjectFile() 实现对对象文件的输出。因为是序列化所以要指定类型。

 

3.8.2 JDBC

详见综合案例

 

3.9 算子综合应用案例

<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.ch</groupId>
    <artifactId>spark_code</artifactId>
    <version>1.0-SNAPSHOT</version>

    <properties>
        <maven.compiler.source>1.8</maven.compiler.source>
        <maven.compiler.target>1.8</maven.compiler.target>
        <scala.version>2.12.10</scala.version>
        <spark.version>2.4.5</spark.version>
        <hadoop.version>2.9.2</hadoop.version>
        <encoding>UTF-8</encoding>
    </properties>

    <dependencies>
        <dependency>
            <groupId>org.scala-lang</groupId>
            <artifactId>scala-library</artifactId>
            <version>${scala.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-core_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-sql_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>

         <!-- 第三方的日期时间包 -->
        <dependency>
            <groupId>joda-time</groupId>
            <artifactId>joda-time</artifactId>
            <version>2.9.7</version>
        </dependency>

        <dependency>
            <groupId>mysql</groupId>
            <artifactId>mysql-connector-java</artifactId>
            <version>5.1.44</version>
        </dependency>

        <dependency>
            <groupId>org.apache.spark</groupId>
            <artifactId>spark-hive_2.12</artifactId>
            <version>${spark.version}</version>
        </dependency>

    </dependencies>
    <build>
        <pluginManagement>
            <plugins>
                <!-- 编译scala的插件 -->
                <plugin>
                    <groupId>net.alchim31.maven</groupId>
                    <artifactId>scala-maven-plugin</artifactId>
                    <version>3.2.2</version>
                </plugin>
                <!-- 编译java的插件 -->
                <plugin>
                    <groupId>org.apache.maven.plugins</groupId>
                    <artifactId>maven-compiler-plugin</artifactId>
                    <version>3.5.1</version>
                </plugin>
            </plugins>
        </pluginManagement>
        <plugins>
            <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <executions>
                    <execution>
                        <id>scala-compile-first</id>
                        <phase>process-resources</phase>
                        <goals>
                            <goal>add-source</goal>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                    <execution>
                        <id>scala-test-compile</id>
                        <phase>process-test-resources</phase>
                        <goals>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <executions>
                    <execution>
                        <phase>compile</phase>
                        <goals>
                            <goal>compile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>
            <!-- 打jar插件 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>2.4.3</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <filters>
                                <filter>
                                    <artifact>*:*</artifact>
                                    <excludes>
                                        <exclude>META-INF/*.SF</exclude>
                                        <exclude>META-INF/*.DSA</exclude>
                                        <exclude>META-INF/*.RSA</exclude>
                                    </excludes>
                                </filter>
                            </filters>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
        </plugins>
    </build>

</project>

1、WordCount - scala

备注:打包上传服务器运行

package com.ch.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object ScalaWordCount {
  def main(args: Array[String]): Unit = {

    // 1、创建SparkContext, 必须要给地址和应用名称
    //val conf = new SparkConf().setMaster("local").setAppName("WordCount")
    // 在集群运行,要用下面的设置
    val conf = new SparkConf().setAppName("WordCount")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 2、读本地文件(集群运行:输入参数)
    //val lines = sc.textFile("data/wc.txt")
    // 集群运行,要用下面的设置
    val lines: RDD[String] = sc.textFile(args(0))

    // 3、RDD转换
    val words: RDD[String] = lines.flatMap(line => line.split("\\s+"))
    val wordsMap: RDD[(String, Int)] = words.map(x => (x, 1))
    val result: RDD[(String, Int)] = wordsMap.reduceByKey(_ + _)

    // 4、输出
    result.foreach(println)

    // 5、关闭SparkContext
    sc.stop()

    // 6、打包,使用spark-submit提交集群运行
//    spark-submit --master local[*] --class com.ch.sparkcore.ScalaWordCount \
//    original-spark_code-1.0-SNAPSHOT.jar /wcinput/*

//    spark-submit --master yarn --class com.ch.sparkcore.ScalaWordCount \
//    original-spark_code-1.0-SNAPSHOT.jar /wcinput/*
  }
}

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

2、WordCount - java

Spark提供了:Scala、Java、Python、R语言的API;
对 Scala 和 Java 语言的支持最好;

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import scala.Tuple2;

import java.util.Arrays;

public class JavaWordCount {
    public static void main(String[] args) {
        // 1 创建 JavaSparkContext
        SparkConf conf = new SparkConf().setAppName("JavaWordCount").setMaster("local[*]");
        JavaSparkContext jsc = new JavaSparkContext(conf);
        jsc.setLogLevel("warn");

        // 2 生成RDD
        JavaRDD<String> lines = jsc.textFile("file:///data\\wc.txt");

        // 3 RDD转换
        JavaRDD<String> words = lines.flatMap(line -> Arrays.stream(line.split("\\s+")).iterator());
        JavaPairRDD<String, Integer> wordsMap = words.mapToPair(word -> new Tuple2<>(word, 1));
        JavaPairRDD<String, Integer> results = wordsMap.reduceByKey((x, y) -> x + y);

        // 4 结果输出
        results.foreach(elem -> System.out.println(elem));

        // 5 关闭SparkContext
        jsc.stop();
    }
}


备注:
Spark入口点:JavaSparkContext
Value-RDD:JavaRDD;key-value RDD:JavaPairRDD
JavaRDD 和 JavaPairRDD转换

  • JavaRDD => JavaPairRDD:通过mapToPair函数

  • JavaPairRDD => JavaRDD:通过map函数转换

lambda表达式使用 ->

 

3、计算圆周率

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

package com.ch.sparkcore

import org.apache.spark.{SparkConf, SparkContext}

import scala.math.random

object SparkPi {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // 如果没有输入分区, 默认分为100个区
    val slices = if (args.length > 0) args(0).toInt else 100
    val N = 100000000

    // 2、生成、转换RDD
    // 生成 0~1 内的坐标
    val n = sc.makeRDD(1 to N, slices)
        .map(idx => {
          val (x, y) = (random, random)
          if (x*x + y*y <= 1) 1 else 0
        }).sum()

    // 3、结果输出
    val pi = 4.0 * n / N
    println(s"pi = $pi")

    // 5、关闭SparkContext
    sc.stop()
  }
}

 

4、广告数据统计

数据格式:timestamp province city userid adid 时间点 省份 城市 用户 广告
需求: 1、统计每一个省份点击TOP3的广告ID 2、统计每一个省份每一个小时的TOP3广告ID

package com.ch.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object Adstat {
  def main(args: Array[String]): Unit = {
    // 1、创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    // 获取前N个
    val N = 3

    // 2、生成RDD
    val lines: RDD[String] = sc.textFile("data\\advert.log")
    // 数据格式:时间点			省份		城市	用户		广告

    // 3、RDD转换

    println(" 1、统计每一个省份点击TOP3的广告ID=========================================================")
    val stat1RDD: RDD[(String, String)] = lines.map { line =>
      val fields: Array[String] = line.split("\\s+")
      (fields(1), fields(4))
    }
    // ((Henan,5),1) ((Henan,5),1) ((Henan,5),1) ....

    // 按省份、广告汇总

    val reduce1RDD: RDD[((String, String), Int)] =
      stat1RDD.map { case (provice, adid) => ((provice, adid), 1) }
        .reduceByKey(_ + _)   // ((Henan,5),2189) ((Hebei,7),2250) ((Henan,0),2237) ....

    // 对以上汇总信息求Top3
    reduce1RDD.map{case ((provice, adid), count) => (provice, (adid, count))}
      .groupByKey()
      .mapValues(buf => buf.toList.sortWith(_._2 > _._2).take(N).map(_._1).mkString(":"))
      .foreach(println)      // (Hebei,7:8:3) (Hunan,5:1:2)  (Hubei,8:6:2)  (Henan,6:0:4)  (Jiangsu,7:3:6)


    println("2、统计每一个省份每一个小时的TOP3广告ID  =========================================================")
    // 0 1 4  时间点		省份		广告
    // 使用field返回需要的数据
    lines.map { line =>
      val fields: Array[String] = line.split("\\s+")
      ((getHour(fields(0)), fields(1), fields(4)), 1)
    }.reduceByKey(_+_)
    // ((0,Hubei,5),2204)
    // ((0,Henan,9),2180)
    // ((0,Jiangsu,0),2147) ......

    lines.map { line =>
      val fields: Array[String] = line.split("\\s+")
      ((getHour(fields(0)), fields(1), fields(4)), 1)
    }.reduceByKey(_+_)
      .map{case ((hour, provice, adid), count) => ((provice, hour), (adid, count))}
    // ((Hubei,0),(5,2204))
    // ((Henan,0),(9,2180))
    // ((Jiangsu,0),(0,2147))
    
    lines.map { line =>
      val fields: Array[String] = line.split("\\s+")
      ((getHour(fields(0)), fields(1), fields(4)), 1)
    }.reduceByKey(_+_)
      .map{case ((hour, provice, adid), count) => ((provice, hour), (adid, count))}
      .groupByKey()
      .mapValues(buf => buf.toList.sortWith(_._2 > _._2).take(N).map(_._1).mkString(":"))
      .collect.foreach(println)
    // ((Henan,0),6:0:4) ((Hubei,0),8:6:2) ((Hunan,0),5:1:2) ((Jiangsu,0),7:3:6) ((Hebei,0),7:8:3)

    // 5、关闭SparkContext
    sc.stop()
  }

  // 下面的功能函数, 必须在外面测试了之后再拿来用
  // 在 spark core 程序中一定不要使用java8以前额时间日期类型, 因为线程不安全
  // 使用第三方的时间日期类型包, 一定要确认其实线程安全的 !!!!
  // 不管知不知道函数的返回值, 最好都要写上返回值
  def getHour(str: String): Int = {
    import org.joda.time.DateTime

    val dt = new DateTime(str.toLong)
    dt.getHourOfDay
  }
}


在Java 8出现前的很长时间内成为Java中日期时间处理的事实标准,用来弥补JDK的不足。
Joda 类具有不可变性,它们的实例无法被修改。(不可变类的一个优点就是它们是线程安全的)
在 Spark Core 程序中使用时间日期类型时,不要使用 Java 8 以前的时间日期类型,线程不安全。

6.2.2 Spark Act触发《KVRDD》创建, 类似map,聚合(重要),排序,JOIN,ACT《输入出》文本,CSV,JSON,Seq,对象,JDBC《算子综合》词数,PAI,广告,共同好友

 

5、找共同好友

原始数据:
100, 200 300 400 500 600
200, 100 300 400
300, 100 200 400 500
400, 100 200 300
500, 100 300
600, 100
第一列表示用户,后面的表示该用户的好友

 

要求:
1、查找两两用户的共同好友
2、最后的结果按前两个id号有序排序

package com.ch.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object FindFriends {
  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    val lines: RDD[String] = sc.textFile("data\\fields.dat")

    println("// 方法一:核心思想利用笛卡尔积求两两的好友,然后去除多余的数据")
    val friendsRDD: RDD[(String, Array[String])] = lines.map { line =>
      val fields: Array[String] = line.split(",")
      val userId = fields(0).trim
      val friends: Array[String] = fields(1).trim.split("\\s+")
      (userId, friends)
    }

    friendsRDD.cartesian(friendsRDD) //.collect().foreach(println)
    //((100,[Ljava.lang.String;@285f38f6),(200,[Ljava.lang.String;@3ab6678b))
    //((200,[Ljava.lang.String;@2b59501e),(100,[Ljava.lang.String;@476e8796))......

    friendsRDD.cartesian(friendsRDD)
      .filter { case ((id1, _), (id2, _)) => id1 < id2 }
    // .collect().foreach(x => println(x._1._1, x._1._2.toBuffer, x._2._1, x._2._2.toBuffer))
    //(100,ArrayBuffer(200, 300, 400, 500, 600),200,ArrayBuffer(100, 300, 400))
    //(100,ArrayBuffer(200, 300, 400, 500, 600),300,ArrayBuffer(100, 200, 400, 500))......

    friendsRDD.cartesian(friendsRDD)
      .filter { case ((id1, _), (id2, _)) => id1 < id2 }
      .map { case ((id1, friends1), (id2, friends2)) =>
        //((id1, id2), friends1.toSet & friends2.toSet)
        ((id1, id2), friends1.intersect(friends2).sorted.toBuffer)
      }.sortByKey()
    // .collect().foreach(println)
    //((100,200),ArrayBuffer(300, 400))
    //((100,300),ArrayBuffer(200, 400, 500))......


    println("// 方法二:消除笛卡尔积,更高效。========================================")
    // 核心思想:将数据变形,找到两两的好友, 再执行数据的合并

    val value: RDD[(String, Array[String])] = friendsRDD.flatMapValues {
      friends => friends.combinations(2)
    }
    //value.collect.foreach(x => println(x._1, x._2.toBuffer))
    //(100,ArrayBuffer(200, 300))
    //(100,ArrayBuffer(200, 400))
    //(100,ArrayBuffer(200, 500))

    friendsRDD.flatMapValues(friends => friends.combinations(2))
      .map(x => (x._2.mkString(" & "), x._1))
     // .collect.foreach(println)
    //(200 & 300,100)
    //(200 & 400,100)
    //(200 & 500,100).....

    println("=====================================================")


    friendsRDD.flatMapValues(friends => friends.combinations(2))
      // .map(x => (x._2.mkString(" & "), Set(x._1)))
      .map { case (k, v) => (v.mkString(" & "), Set(k)) }
      .reduceByKey(_ | _)
      .sortByKey()
      .collect().foreach(println)
    //(100 & 200,Set(300, 400))
    //(100 & 300,Set(200, 400, 500)).....

    // 知识点:flatMapValues / combinations / 数据的变形 / reduceByKey / 集合的操作
    // combinations 求的是数组中的排列
    // 关闭SparkContext
    sc.stop()
  }
}
val s1 = (1 to 5).toSet
val s2 = (3 to 8).toSet

// 交。intersect
println(s1 & s2)

// 并。union
println(s1 | s2)

// 差。diff
println(s1 &~ s2)

 

6、Super WordCount

要求:将单词全部转换为小写,去除标点符号(难),去除停用词(难);最后按照 count 值降序保存到文件,同时将全部结果保存到MySQL(难);标点符号和停用词可以自定义。
停用词:语言中包含很多功能词。与其他词相比,功能词没有什么实际含义。最普遍的功能词是[限定词](the、a、an、that、those),介词(on、in、to、from、over等)、代词、数量词等。
Array[(String, Int)] => scala jdbc => MySQL

package com.ch.sparkcore

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SuperWordCount1 {
  private val stopWords = "in on to from by a an the is are were was i we you your he his some any of as can it each".split("\\s+")
  private val punctuation = "[\\)\\.,:;'!\\?]"

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")
    val acc3 = sc.collectionAccumulator[String]("allWords")

    // RDD转换
    // 换为小写,去除标点符号(难),去除停用词(难)
    val lines: RDD[String] = sc.textFile("data\\swc.dat")
    lines.flatMap(_.split("\\s+"))
      .map(_.toLowerCase)
      .map(_.replaceAll(punctuation, ""))
      .filter(word => !stopWords.contains(word) && word.trim.length>0)
      .map((_, 1))
      .reduceByKey(_+_)
      .sortBy(_._2, false)
      .collect.foreach(println)

    // 结果输出

    // 关闭SparkContext
    sc.stop()
  }
}


 

create table wordcount(word varchar(30), count int);
package com.ch.sparkcore

import java.sql.{Connection, DriverManager, PreparedStatement}

object JDBCDemo {
  def main(args: Array[String]): Unit = {

    // 定义结果集Array[(String, Int)]
    val str = "hadoop spark java scala hbase hive sqoop hue tez atlas datax grinffin zk kafka"
    val result: Array[(String, Int)] = str.split("\\s+").zipWithIndex

    // 定义参数
    val url = "jdbc:mysql://linux123:3306/ebiz?useUnicode=true&characterEncoding=utf-8&useSSL=false"
    val username = "hive"
    val password = "12345678"

    // jdbc 保存数据
    var conn: Connection = null
    var stmt: PreparedStatement = null
    val sql = "insert into wordcount values (?, ?)"
    try {
      conn = DriverManager.getConnection(url, username, password)
      stmt = conn.prepareStatement(sql)
      result.foreach{case (k, v) =>
        stmt.setString(1, k)
        stmt.setInt(2, v)
        stmt.executeUpdate()
      }
    }catch {
      case e: Exception => e.printStackTrace()
    } finally {
      if (stmt != null) stmt.close()
      if (conn != null) conn.close()
    }
  }
}


未优化的程序:使用 foreach 保存数据,要创建大量的链接

package com.ch.sparkcore

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SuperWordCount2 {
  private val stopWords = "in on to from by a an the is are were was i we you your he his some any of as can it each".split("\\s+")
  private val punctuation = "[\\)\\.,:;'!\\?]"
  private val url = "jdbc:mysql://linux123:3306/ebiz?useUnicode=true&characterEncoding=utf-8&useSSL=false"
  private val username = "hive"
  private val password = "12345678"

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // RDD转换
    // 换为小写,去除标点符号(难),去除停用词(难)
    val lines: RDD[String] = sc.textFile("data\\swc.dat")
    val resultRDD: RDD[(String, Int)] = lines.flatMap(_.split("\\s+"))
      .map(_.toLowerCase)
      .map(_.replaceAll(punctuation, ""))
      .filter(word => !stopWords.contains(word) && word.trim.length > 0)
      .map((_, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)

    // 结果输出
    resultRDD.saveAsTextFile("data\\superwc")
    // 使用 foreach,对每条记录创建连接
    resultRDD.foreach{case (k, v) =>
      var conn: Connection = null
      var stmt: PreparedStatement = null
      val sql = "insert into wordcount values (?, ?)"
      try {
        conn = DriverManager.getConnection(url, username, password)
        stmt = conn.prepareStatement(sql)
        stmt.setString(1, k)
        stmt.setInt(2, v)
        stmt.executeUpdate()
      }catch {
        case e: Exception => e.printStackTrace()
      } finally {
        if (stmt != null) stmt.close()
        if (conn != null) conn.close()
      }
    }

    // 关闭SparkContext
    sc.stop()
  }
}


优化后的程序:使用 foreachPartition 保存数据,一个分区创建一个链接;cache RDD

package com.ch.sparkcore

import java.sql.{Connection, DriverManager, PreparedStatement}

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

object SuperWordCount3 {
  private val stopWords: Array[String] = "in on to from by a an the is are were was i we you your he his some any of as can it each".split("\\s+")
  private val punctuation = "[\\)\\.,:;'!\\?]"
  private val url = "jdbc:mysql://linux123:3306/ebiz?useUnicode=true&characterEncoding=utf-8&useSSL=false"
  private val username = "hive"
  private val password = "12345678"

  def main(args: Array[String]): Unit = {
    // 创建SparkContext
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init).setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    // RDD转换
    // 换为小写,去除标点符号(难),去除停用词(难)
    val lines: RDD[String] = sc.textFile("file:///C:\\Project\\LagouBigData\\data\\swc.dat")
    val resultRDD: RDD[(String, Int)] = lines.flatMap(_.split("\\s+"))
      .map(_.toLowerCase)
      .map(_.replaceAll(punctuation, ""))
      .filter(word => !stopWords.contains(word) && word.trim.length > 0)
      .map((_, 1))
      .reduceByKey(_ + _)
      .sortBy(_._2, false)

    //resultRDD.foreach(println)
    // ?????这里是干嘛的?
    resultRDD.cache()

    // 结果输出
    resultRDD.saveAsTextFile("file:///C:\\Project\\LagouBigData\\data\\superwc")
    // 使用 foreachPartition,对每条记录创建连接
    resultRDD.foreachPartition { iter => saveAsMySQL(iter)}

    // 关闭SparkContext
    sc.stop()
  }

  def saveAsMySQL(iter: Iterator[(String, Int)]): Unit = {
    var conn: Connection = null
    var stmt: PreparedStatement = null
    val sql = "insert into wordcount values (?, ?)"

    try {
      conn = DriverManager.getConnection(url, username, password)
      stmt = conn.prepareStatement(sql)
      iter.foreach { case (k, v) =>
        stmt.setString(1, k)
        stmt.setInt(2, v)
        stmt.executeUpdate()
      }
    } catch {
      case e: Exception => e.printStackTrace()
    } finally {
      if (stmt != null) stmt.close()
      if (conn != null) conn.close()
    }
  }
}
// 在SparkSQL中有内建的访问MySQL的方法,调用非常方便
// SparkCore、SQL不支持的外部存储, 可以使用上述方法


备注:

  • SparkSQL有方便的读写MySQL的方法,给参数直接调用即可;

  • 但以上掌握以上方法非常有必要,因为SparkSQL不是支持所有的类型的数据库

上一篇:【转:学习】「2020」PDD数据分析笔试题 | 附解答


下一篇:activiti知识点梳理