依赖hadoop
windows
详见下面链接：
https://www.malaoshi.top/show_1IXa3fALldq.html
idea 安装 scala 插件
https://malaoshi.top/show_1IXnyD53SzF.html
创建工程
添加 scala 框架支持
创建scala源码文件夹
标记文件夹为源码
修改 idea maven
修改 maven 路径
必须按下图勾选，否则 使用 aggregateByKey() 算子时报错
错误如下：
org.apache.spark.SparkException: Task not serializable
准备数据文件
在 工程 下创建 data 文件夹
在该文件夹下创建 wordcount.txt 文件，内容如下：
李雷 lucy
韩梅梅 李雷
lucy 韩梅梅
lili 李雷
张三 lucy
李四 韩梅梅
王五 韩梅梅
代码
pom.xml
<dependencies>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-core_2.12</artifactId>
<version>3.0.0</version>
</dependency>
<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-yarn_2.12</artifactId>
<version>3.0.0</version>
</dependency>
</dependencies>
<build>
<sourceDirectory>src/main/scala</sourceDirectory>
<plugins>
<plugin>
<groupId>net.alchim31.maven</groupId>
<artifactId>scala-maven-plugin</artifactId>
<version>3.3.2</version>
<executions>
<execution>
<goals>
<goal>compile</goal>
<goal>testCompile</goal>
</goals>
</execution>
</executions>
</plugin>
</plugins>
</build>
log4j.properties（可略）
log4j.rootLogger=WARN, stdout
log4j.appender.stdout=org.apache.log4j.ConsoleAppender
log4j.appender.stdout.layout=org.apache.log4j.PatternLayout
log4j.appender.stdout.layout.ConversionPattern=%d %p [%c] - %m%n
创建scala文件
实现方式1
类似 scala的方式，详见 链接
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object Test {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)
// 1. 读取文件，获取一行一行的数据。由于是 local 本地模式，所以默认读取本机文件
val rdd: RDD[String] = sc.textFile("data/wordcount.txt")
// 2. 将一行数据进行拆分，形成一个一个的单词（分词）
// 扁平化：将整体拆分成个体的操作
// "hello world" => hello, world, hello, world
val rdd2: RDD[String] = rdd.flatMap(_.split(" "))
//分组
val rdd3: RDD[(String, Iterable[String])] = rdd2.groupBy(e => e)
//统计value的长度
val resRDD: RDD[String] = rdd3.map(e => s"${e._1},${e._2.size}")
resRDD.collect().foreach(println)
resRDD.saveAsTextFile("res")
sc.stop()
}
}
警告：移除本地运行模式
上面的代码中，应该移除 .setMaster("local[*]")，否则打包发布到yarn运行时，会报错
移除后，需要做下面配置，否则运行失败：
在该文件 点右键，如按下图点击：
设置本地运行模式，如下图：
在下图红框处输入：-Dspark.master=local[*]，表示本地运行模式，如下图
实现方式2
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)
// 1. 读取文件，获取一行一行的数据。由于是 local 本地模式，所以默认读取本机文件
val rdd:RDD[String] = sc.textFile("data/wordcount.txt")
// 2. 将一行数据进行拆分，形成一个一个的单词（分词）
// 扁平化：将整体拆分成个体的操作
// "hello world" => hello, world, hello, world
val rdd2: RDD[String] = rdd.flatMap(_.split(" "))
//统计各个元素出现的次数
val res: collection.Map[String, Long] = words.countByValue()
res.foreach(println)
sc.stop()
实现方式3
package test222
import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}
object WordCount {
def main(args: Array[String]): Unit = {
// 建立和Spark框架的连接
val sparConf = new SparkConf().setMaster("local[*]").setAppName("WordCount")
val sc = new SparkContext(sparConf)
// TODO 执行业务操作
// 1. 读取文件，获取一行一行的数据。由于是 local 本地模式，所以默认读取本机文件
val lines: RDD[String] = sc.textFile("data/1.txt")
println("lines============================")
lines.foreach(println)
// 2. 将一行数据进行拆分，形成一个一个的单词（分词）
// 扁平化：将整体拆分成个体的操作
// "hello world" => hello, world, hello, world
val words: RDD[String] = lines.flatMap(_.split(" "))
println("words============================")
words.foreach(println)
// 3. 在单词拆分的基础上对每个单词实例计数为1, 也就是 word => (word, 1)
// val pairs = words.map(word => (word, 1))
// val pairs = words.map(word => (_, 1))
val wordOne: RDD[(String, Int)] = words.map((_, 1))
println("wordOne============================")
wordOne.foreach(println)
// 4. 在每个单词实例计数为1的基础之上统计每个单词在文件中出现的总次数, 即key相同的value相加
// val wordscount = wordOne.reduceByKey((v1, v2) => v1 + v2)
val wordscount: RDD[(String,Int)] = wordOne.reduceByKey(_ + _)
println("wordscount============================")
wordscount.foreach(println)
// 5. 将数据聚合结果采集到内存中。用collect会将集群中的数据收集到当前运行drive的机器上，需要保证单台机器能放得下所有数据
val res: Array[(String, Int)] = wordscount.collect()
// 6. 打印结果
println("wordscount============================")
res.foreach(println)
// 7. 关闭连接
sc.stop()
}
}